import sys, os from tools.Get_Json import get_defect_detail from tools.defines import * def caculate_work_days(start_date : str, end_date : str) -> str: """根据起止日期计算工期 Args: start_date (str): 格式: yyyy-mm-ddThh:mm:ss end_date (str): 格式: yyyy-mm-ddThh:mm:ss Returns: str: 计算出来的总工期，单位为天 """ import datetime if start_date == None or end_date == None: return f"开始时间{start_date} ---- 结束时间{end_date} 日期格式错误" start_date = datetime.datetime.strptime(start_date, '%Y-%m-%dT%H:%M:%S') end_date = datetime.datetime.strptime(end_date, '%Y-%m-%dT%H:%M:%S') return (end_date - start_date).days def get_year_month(date): """根据格式化date字符串获取年月 'date': '二〇二一年十二月十日 9:00' Args: date (str): 日期字符串 Returns: 年月字符串 '二〇二一年十二月' """ unit_map = {'1' : '一', '2' : '二', '3' : '三', '4' : '四', '5' : '五', '6' : '六', '7' : '七', '8' : '八', '9' : '九', '0' : '〇'} unit_map_month = {1 : '一', 2 : '二', 3 : '三', 4 : '四', 5 : '五', 6 : '六', 7 : '七', 8 : '八', 9 : '九', 10 : '十', 11 : '十一', 12 : '十二'} year = date.split('年')[0] month = date.split('年')[1].split('月')[0] year = ''.join([unit_map[i] for i in year]) month = unit_map_month[int(month)] return f"{year}年{month}月" def merge_info(frontend_info, default_info): """ 合并前端传入的 info 和默认 info 新规则：以default_info为基础字典，用frontend_info完全覆写，取两者的并集 Args: frontend_info: 前端传入的字典 default_info: 默认的完整字典 Returns: 合并后的完整字典 """ if not isinstance(frontend_info, dict) or frontend_info is None: return default_info.copy() # 先复制默认字典 merged_info = default_info.copy() # 用前端字典完全覆写 merged_info.update(frontend_info) return merged_info def merge_dicts(dict1, dict2): # 创建一个新的字典来存储合并结果 merged_dict = {} # 遍历第一个字典 for key, value_list in dict1.items(): if key in dict2: # 如果键在第二个字典中存在，合并两个列表 merged_dict[key] = value_list + dict2[key] else: # 如果键在第二个字典中不存在，直接使用第一个字典的值列表 merged_dict[key] = value_list # 遍历第二个字典 for key, value_list in dict2.items(): if key not in dict1: # 如果键在第一个字典中不存在，直接使用第二个字典的值列表 merged_dict[key] = value_list return merged_dict def get_defect_str(Y_defect_list : list[dict] ) -> list: """将叶片缺陷信息转换为一条条描述信息 Args: Y_defect_list (list):[ { 'record' : {'defectId': '02d892a178a82561bb565559102c7a58', 'imageId': '41543f531be24522b7bec741b9a483a2', 'defectName': '手动添加的缺陷1', 'defectCode': None, 'partName': '叶片1', 'defectTypeLabel': '表面裂纹', 'defectType': 'bmlw', 'defectLevelLabel': '轻微缺陷', 'defectLevel': 'SLIGHT', 'defectPosition': '', 'description': '手动添加的缺陷，请填写详细描述', 'repairIdea': '建议进行进一步检查', 'labelInfo': None, 'markInfo': {'label': None, 'clsId': None, 'bbox': None, 'confidence': 0.0}} 'imagePath' : '/image/path' }, .... ] Returns: result (list): [ "叶片1表面裂纹轻微缺陷，位于{defectPosition},建议进行进一步检查", ], ... """ result = [] for item in Y_defect_list: record = item['record'] defect_type_label = record.get('defectTypeLabel', '未知类型') defect_level_label = record.get('defectLevelLabel', '未知等级') defect_position = record.get('defectPosition', '未知位置') repair_idea = record.get('repairIdea', '无建议') defect_description = f"{defect_type_label}{defect_level_label}缺陷，位于{defect_position}，{repair_idea}" result.append(defect_description) return result def safe_get(data, *keys, default=None): """ 递归安全访问嵌套字典的键，如果中间键不存在，则返回默认值 Args: data (dict): 要访问的字典 *keys: 要访问的键（可以是多个，如 "叶片1", "裂纹"） default: 如果键不存在，返回的默认值（默认None） Returns: 如果所有键都存在，返回对应的值；否则返回 default """ if not keys or data is None: return data if data is not None else default current_key = keys[0] if isinstance(data, dict): return safe_get(data.get(current_key), *keys[1:], default=default) else: return default def is_frozen(): return hasattr(sys, 'frozen') or hasattr(sys, '_MEIPASS') def get_resource_path(relative_path): """ 获取资源的路径 """ if is_frozen(): base_path = sys._MEIPASS else: base_path = os.path.abspath(".") return os.path.join(base_path, relative_path) def get_defedct_info(defect_dict): """获取按缺陷类型分类的缺陷信息 Args: defect_dict (dict): {'叶片1': {'DEFECT':[{'record': { 'defectId': '02d892a178a82561bb565559102c7a58', 'imageId': '41543f531be24522b7bec741b9a483a2', 'defectName': '手动添加的缺陷1', 'defectCode': None, 'partName': '叶片1', 'defectTypeLabel': '表面裂纹', 'defectType': 'bmlw', 'defectLevelLabel': '轻微缺陷', 'defectLevel': 'SLIGHT', 'defectPosition': '', 'description': '手动添加的缺陷，请填写详细描述', 'repairIdea': '建议进行进一步检查', 'labelInfo': None, 'markInfo': {'label': None, 'clsId': None, 'bbox': None, 'confidence': 0.0}, 'imagePath': '/static/image/temp/out-work/12bc30fb209f3af3bf530541c5b062bd/F02_机组10060_叶片，叶片PS面距叶根15.5米处轴向裂纹，轴向：260mm，弦向：20mm。DJI_20250526090227_0052_Z.png' }, ... },] } }, ... Returns: result (dict): { '叶片1': { #按叶片分记录 '表面裂纹': { 'SLIGHT': #按类型分等级,等级从'defectLevel'获取 [ #按等级分记录 { record:{...} }, #一张图片一个记录 ... ] ... } } } result2 (dict): { '叶片1': { #按叶片分记录 '表面裂纹': [ #按类型分记录 """ result = {} #缺陷等级数量列表 result2 = {} #缺陷树状列表 for part_name, part_info in defect_dict.items(): if part_name not in result: result[part_name] = {} result2[part_name] = {} for defect_type, defect_list in part_info.items(): for defect_dict in defect_list: record = defect_dict['record'] type_label = record['defectTypeLabel'] level = record['defectLevel'] if type_label not in result[part_name]: result2[part_name][type_label] = [] if type_label not in result[part_name]: result[part_name][type_label] = {} if level not in result[part_name][type_label]: result[part_name][type_label][level] = [] result[part_name][type_label][level].append(defect_dict) p = record['defectPosition'] t = record['defectTypeLabel'] l = record['defectLevelLabel'] d = record['description'] did = record['defectId'] iid = record['imageId'] detail = get_defect_detail(did) a = detail.get('axial', '') c = detail.get('chordwise', '') area = f"轴向：{a}mm，弦向：{c}mm" record_info = [f"{p}，{t}，{l}，{d}", area, iid] result2[part_name][type_label].append(record_info) return result, result2 def get_defect_json(defect_info: dict, Y_code: list[str], turbine_code: str) -> dict: """将缺陷信息转换为表格形式的JSON格式 Args: defect_info: 结构为 {叶片1/2/3: {缺陷类型: [{'record': {...}}, ...]}} Y_code: 叶片编号列表，如 ['1001', '1002', '1003'] turbine_code: 机组编号，如 'G01' Returns: { "type": "table", "content": { "rows": 总行数, "cols": 4, "merged_cells": [合并信息], "cells": [表格内容] } } """ # 初始化表格 cells = [] merged_cells = [] # 默认设置 default_border = { "top": {"style": "single", "size": "4", "color": "000000"}, "left": {"style": "single", "size": "4", "color": "000000"}, "bottom": {"style": "single", "size": "4", "color": "000000"}, "right": {"style": "single", "size": "4", "color": "000000"} } # 1. 计算总行数（先计算再创建表头） total_defect_rows = sum(len(defects) for blade_defects in defect_info.values() for defects in blade_defects.values()) total_rows = 1 + total_defect_rows # 表头 + 数据行 # 2. 创建表头行（第0行） header_row = [ create_cell(0, 0, "机组号", default_border), create_cell(0, 1, "叶片号", default_border), create_cell(0, 2, "损坏描述", default_border), create_cell(0, 3, "缺陷面积", default_border) ] cells.append(header_row) # 3. 填充数据行（从第1行开始） current_data_row = 1 # 合并机组号（如果有多行数据） if total_defect_rows > 1: merged_cells.append({ "start_row": 1, # 第2行(0-based中的1) "start_col": 0, "end_row": total_rows - 1, # 最后一行(0-based) "end_col": 0 }) # 填充机组号（仅第一行数据行） if total_defect_rows > 0: first_data_row = [ create_cell(1, 0, turbine_code, default_border, is_merged=total_defect_rows>1, is_primary=True), None, None, None ] cells.append(first_data_row) # 按叶片顺序填充数据 for blade_idx, (blade_key, blade_defects) in enumerate(defect_info.items(), 1): blade_code = Y_code[blade_idx - 1] blade_defect_count = sum(len(defects) for defects in blade_defects.values()) if blade_defect_count > 1: merged_cells.append({ "start_row": current_data_row, # 当前行(0-based) "start_col": 1, "end_row": current_data_row + blade_defect_count - 1, # 结束行(0-based) "end_col": 1 }) # 填充缺陷数据 for defect_type, records in blade_defects.items(): for record in records: # 确保不超过总行数 if current_row >= total_rows: break description = record.get('record', {}).get('损坏描述', '待填写') area = record.get('record', {}).get('缺陷面积', '待填写') is_primary = (not merged_cells or current_row == merged_cells[-1]["start_row"]) row_data = [ None, # 机组号(已合并) create_cell(current_row, 1, blade_code, default_border, is_merged=blade_defect_count>1, is_primary=is_primary), create_cell(current_row, 2, str(description), default_border), create_cell(current_row, 3, str(area), default_border) ] cells.append(row_data) current_row += 1 # 填充空单元格 for row_idx, row in enumerate(cells): for col_idx in range(4): if row[col_idx] is None: row[col_idx] = create_cell(row_idx, col_idx, "", default_border) # 最终行数检查（调试用） print(f"调试信息: 预期行数={total_rows}, 实际行数={len(cells)}") print(f"缺陷信息统计: 叶片数={len(defect_info)}, 总缺陷数={total_defect_rows}") return [{ "type": "table", "content": { "rows": len(cells), # 使用实际行数而非计算行数 "cols": 4, "merged_cells": merged_cells, "cells": cells } }] def create_cell(row, col, text, border, is_merged=False, is_primary=False): """创建标准化的单元格结构""" return { "row": row, "col": col, "is_merged": is_merged, "alignment": "center", "border": border, "merge_info": {"is_primary": is_primary}, "content": [{ "alignment": "center", "runs": [{ "text": text, "font": { "name": "宋体", "size": 12, "bold": False, "italic": False, "underline": False, "color": {"r": 0, "g": 0, "b": 0} } }] }] } from typing import Dict, List, Any def tree_dict_to_table_data(tree: Dict) -> List[List[Any]]: """ 将树状字典转换为二维表格数据（保留原始类型） 参数: tree: 树状字典，结构为dict[dict[...[list]]] 返回: 二维列表，每个子列表代表一条路径（元素类型与输入一致），确保最内层列表被展开 """ result = [] def traverse(node: Dict, path: List[Any]) -> None: for key, value in node.items(): current_path = path + [key] # 保留键的原始类型 if isinstance(value, dict): traverse(value, current_path) elif isinstance(value, list): for item in value: if isinstance(item, list): # 如果列表项本身是列表，则展开它 result.append(current_path + item) else: # 否则作为单个元素添加 result.append(current_path + [item]) else: result.append(current_path + [value]) # 保留值的原始类型 traverse(tree, []) return result def defect_list_addtitle(list_data: List[List[str]], turbine_code: str): """ 给缺陷列表添加标题行 """ list_data.insert(0, DEFECT_TABLE) for i,l1 in enumerate(list_data): if i == 0: continue l1.insert(0, turbine_code) return list_data