async def get_font_style(cell): """获取单元格的字体样式信息""" font = cell.font style = [] if font.bold: style.append("font-weight: bold;") if font.italic: style.append("font-style: italic;") if font.underline: style.append("text-decoration: underline;") if font.color and font.color.rgb: style.append(f"color: #{font.color.rgb};") if font.size: style.append(f"font-size: {font.size}pt;") return " ".join(style) async def excel_to_html_with_merges(nss={}): try: update_id = nss.get('update_id') excel_file = file_realpath(nss.get('filepath')) # excel_file = 'C:/Users/Admin/Desktop/算力业务供应商评估表-模版.xlsx' workbook = openpyxl.load_workbook(filename=excel_file, data_only=True) sheet = workbook.active basic_info_init = [ {"id": "1", "label": "算力型号", "type": "select", "options": ["4090", "A100", "A800", "H100", "H800", "A100-40G", "A800-40G", "A100-PCIE-40G", "A100-PCIE-80G", "A800-PCIE-40G", "A800-PCIE-80G"], "value": sheet['B3'].value, "is_new": 0}, {"id": "2", "label": "计算网络类型", "type": "select", "options": ["IB", "ROCE", "以太"], "value": sheet['F3'].value, "is_new": 0}, {"id": "3", "label": "资源类型", "type": "select", "options": ["可控", "临时"], "value": sheet['J3'].value, "is_new": 0}, {"id": "4", "label": "集群物理位置", "type": "input", "value": sheet['N3'].value, "is_new": 0}, {"id": "5", "label": "算力提供方式", "type": "select", "options": ["裸金属", "容器化"], "value": sheet['B4'].value, "is_new": 0}, {"id": "6", "label": "剩余GPU数量", "type": "input", "value": sheet['F4'].value, "is_new": 0}, {"id": "7", "label": "是否现货", "type": "input", "value": sheet['J4'].value, "is_new": 0}, {"id": "8", "label": "资源入库时间", "type": "input", "value": sheet['N4'].value, "is_new": 0}, {"id": "9", "label": "起始租期", "type": "input", "value": sheet['B5'].value, "is_new": 0}, {"id": "10", "label": "付款方式", "type": "input", "value": sheet['F5'].value, "is_new": 0}, {"id": "11", "label": "月租(建议价)", "type": "input", "value": sheet['J5'].value, "is_new": 0}, {"id": "12", "label": "交付周期", "type": "input", "value": sheet['N5'].value, "is_new": 0}, {"id": "13", "label": "基本配置", "type": "input", "value": sheet['B6'].value, "is_new": 0}, {"id": "14", "label": "集群GPU总量", "type": "input", "value": sheet['B7'].value, "is_new": 0}, {"id": "15", "label": "是否有配套存储", "type": "select", "options": ["有", "无"], "value": sheet['F7'].value, "is_new": 0}, {"id": "16", "label": "存储类型", "type": "select", "options": ["全闪", "混闪"], "value": sheet['J7'].value, "is_new": 0}, {"id": "17", "label": "存储总量", "type": "input", "value": sheet['N7'].value, "is_new": 0}, {"id": "18", "label": "剩余存储数量", "type": "input", "value": sheet['B8'].value, "is_new": 0}, {"id": "19", "label": "有无云平台", "type": "select", "options": ["有", "无"], "value": sheet['F8'].value, "is_new": 0}, {"id": "20", "label": "月租(成本)", "type": "input", "value": sheet['J8'].value, "is_new": 0}, {"id": "21", "label": "集群是否支持改造", "type": "select", "options": ["支持", "不支持"], "value": sheet['N8'].value, "is_new": 0}, {"id": "22", "label": "是否支持测试", "type": "select", "options": ["支持", "不支持"], "value": sheet['B9'].value, "is_new": 0}, {"id": "23", "label": "最大测试台数", "type": "input", "value": sheet['F9'].value, "is_new": 0}, {"id": "24", "label": "是否已经测试验证", "type": "select", "options": ["是", "否"], "value": sheet['J9'].value, "is_new": 0} ] index_id = update_id if update_id else uuid() db = DBPools() async with db.sqlorContext('kboss') as sor: ns_dic = { 'id': index_id, 'orgid': nss.get('orgid'), 'basic_info': json.dumps(basic_info_init) if not isinstance(basic_info_init, str) else basic_info_init, 'remark': nss.get('remark'), 'del_flg': '0' } if update_id: await sor.U('kyy_supply', ns_dic) else: await sor.C('kyy_supply', ns_dic) # 加载工作簿和活动工作表 # wb = openpyxl.load_workbook(excel_file) # sheet = wb.active # 获取合并单元格的信息 merged_ranges = sheet.merged_cells.ranges html_content = '' # 逐行处理 for row in sheet.iter_rows(): html_content += '' for cell in row: # 获取字体样式 font_style = await get_font_style(cell) # 获取当前单元格的行列数 cell_row, cell_col = cell.row, cell.column # 检查当前单元格是否是合并单元格的起始单元格 if any((cell_row, cell_col) == (merge.min_row, merge.min_col) for merge in merged_ranges): # 获取合并单元格的范围 for merge in merged_ranges: if (cell_row, cell_col) == (merge.min_row, merge.min_col): rowspan = merge.max_row - merge.min_row + 1 colspan = merge.max_col - merge.min_col + 1 cell_value = cell.value if cell.value is not None else "" cell_value = str(cell_value).replace('\n', '
') html_content += f'' break elif not any((cell_row, cell_col) in [(merge.min_row + r, merge.min_col + c) for r in range(merge.max_row - merge.min_row + 1) for c in range(merge.max_col - merge.min_col + 1)] for merge in merged_ranges): # 不是合并单元格的后续单元格 cell_value = cell.value if cell.value is not None else "" html_content += f'' html_content += '' html_content += '
{cell_value}{cell_value}
' html_content = html_content.replace("color: #Values must be of type ; ", '') await sor.U('kyy_supply', {'id': index_id, 'excel_info': html_content}) return { 'status': True, 'msg': 'excel覆盖填写信息成功, 上传文档成功' } except Exception as e: return { 'status': False, 'msg': 'excel覆盖填写信息报错, 上传文档报错, %s' % str(e) } ret = await excel_to_html_with_merges(params_kw) return ret