659 lines
20 KiB
Python
659 lines
20 KiB
Python
|
||
from collections import defaultdict
|
||
import re
|
||
from typing import Optional
|
||
import traceback
|
||
import logging
|
||
import pandas as pd
|
||
from openpyxl import load_workbook
|
||
import json
|
||
import os
|
||
# import xlrd
|
||
# =====================================================
|
||
# 行类型判断
|
||
# =====================================================
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def is_data_row(row):
|
||
"""
|
||
判断是否为数据行
|
||
数据行的特征：第1列包含数字
|
||
|
||
Args:
|
||
row (dict): 包含行数据的字典，格式为 {"content": {"列号": "值", ...}}
|
||
|
||
Returns:
|
||
bool: 如果是数据行返回True，否则返回False
|
||
"""
|
||
try:
|
||
v = row["content"].get("1", "")
|
||
result = isinstance(v, str) and v.strip().isdigit()
|
||
logger.debug(f"is_data_row: value='{v}', result={result}")
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"Error in is_data_row: {e}, row: {row}")
|
||
return False
|
||
|
||
|
||
def is_field_row(row):
|
||
"""
|
||
判断是否为字段行
|
||
字段行的特征：至少有5个值，且超过90%的值不是数字，根据环境处理
|
||
|
||
Args:
|
||
row (dict): 包含行数据的字典
|
||
|
||
Returns:
|
||
bool: 如果是字段行返回True，否则返回False
|
||
"""
|
||
try:
|
||
values = list(row["content"].values())
|
||
if len(values) < 5:
|
||
logger.debug(f"is_field_row: less than 5 values, returning False")
|
||
return False
|
||
non_digit = sum(1 for v in values if isinstance(v, str) and not v.strip().isdigit())
|
||
result = non_digit / len(values) > 0.9
|
||
logger.debug(f"is_field_row: non_digit_ratio={non_digit / len(values)}, result={result}")
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"Error in is_field_row: {e}, row: {row}")
|
||
return False
|
||
|
||
|
||
def is_header_row(row):
|
||
"""
|
||
判断是否为表头行
|
||
表头行的特征：列数不超过4个，且不是字段行或数据行，根据环境处理
|
||
|
||
Args:
|
||
row (dict): 包含行数据的字典
|
||
|
||
Returns:
|
||
bool: 如果是表头行返回True，否则返回False
|
||
"""
|
||
try:
|
||
content_len = len(row["content"])
|
||
field_result = is_field_row(row)
|
||
data_result = is_data_row(row)
|
||
|
||
result = content_len <= 4 and not field_result and not data_result
|
||
logger.debug(f"is_header_row: content_len={content_len}, is_field={field_result}, is_data={data_result}, result={result}")
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"Error in is_header_row: {e}, row: {row}")
|
||
return False
|
||
|
||
# =====================================================
|
||
# 多级表头
|
||
# =====================================================
|
||
|
||
def collect_field_rows(rows, start_idx):
|
||
"""
|
||
从指定索引开始收集连续的字段行
|
||
|
||
Args:
|
||
rows (list): 行数据列表
|
||
start_idx (int): 开始收集的索引
|
||
|
||
Returns:
|
||
tuple: (字段行列表, 下一个非字段行的索引)
|
||
"""
|
||
try:
|
||
field_rows = []
|
||
i = start_idx
|
||
initial_i = start_idx
|
||
while i < len(rows) and is_field_row(rows[i]):
|
||
field_rows.append(rows[i]["content"])
|
||
i += 1
|
||
logger.info(f"collect_field_rows: collected {len(field_rows)} field rows from index {initial_i} to {i}")
|
||
return field_rows, i
|
||
except Exception as e:
|
||
logger.error(f"Error in collect_field_rows: {e}")
|
||
return [], start_idx
|
||
|
||
def dedup_path(path):
|
||
"""
|
||
去除路径中的重复元素，保持顺序
|
||
|
||
Args:
|
||
path (list): 路径列表
|
||
|
||
Returns:
|
||
list: 去重后的路径列表
|
||
"""
|
||
try:
|
||
new_path = []
|
||
for p in path:
|
||
if p and (not new_path or new_path[-1] != p):
|
||
new_path.append(p)
|
||
logger.debug(f"dedup_path: original={path}, deduplicated={new_path}")
|
||
return new_path
|
||
except Exception as e:
|
||
logger.error(f"Error in dedup_path: {e}, path: {path}")
|
||
return []
|
||
|
||
def build_field_paths(field_rows):
|
||
"""
|
||
构建字段路径映射，将列号映射到字段路径
|
||
|
||
Args:
|
||
field_rows (list): 字段行列表
|
||
|
||
Returns:
|
||
defaultdict: 列号到字段路径的映射
|
||
"""
|
||
try:
|
||
field_paths = defaultdict(list)
|
||
for row_idx, row in enumerate(field_rows):
|
||
for col, name in row.items():
|
||
if name:
|
||
field_paths[col].append(name)
|
||
|
||
for col in field_paths:
|
||
field_paths[col] = dedup_path(field_paths[col])
|
||
|
||
logger.info(f"build_field_paths: processed {len(field_paths)} columns from {len(field_rows)} rows")
|
||
return field_paths
|
||
except Exception as e:
|
||
logger.error(f"Error in build_field_paths: {e}")
|
||
return defaultdict(list)
|
||
|
||
|
||
# =====================================================
|
||
# 值 & 结构处理
|
||
# =====================================================
|
||
|
||
def normalize_value(value):
|
||
"""
|
||
标准化值，将特殊值转换为对应格式
|
||
|
||
Args:
|
||
value: 待处理的值
|
||
|
||
Returns:
|
||
处理后的值
|
||
"""
|
||
try:
|
||
if value == "√":
|
||
logger.debug(f"normalize_value: converting '√' to True")
|
||
return True
|
||
if value in ("", None):
|
||
logger.debug(f"normalize_value: empty value converted to empty string")
|
||
return ""
|
||
logger.debug(f"normalize_value: value={value} unchanged")
|
||
return value
|
||
except Exception as e:
|
||
logger.error(f"Error in normalize_value: {e}, value: {value}")
|
||
return ""
|
||
|
||
|
||
def insert_by_path(target, path, value):
|
||
"""
|
||
根据路径在目标对象中插入值
|
||
|
||
Args:
|
||
target (dict): 目标字典
|
||
path (list): 插入路径
|
||
value: 待插入的值
|
||
"""
|
||
try:
|
||
if not path:
|
||
logger.warning("insert_by_path: empty path provided")
|
||
return
|
||
|
||
cur = target
|
||
for key in path[:-1]:
|
||
cur = cur.setdefault(key, {})
|
||
cur[path[-1]] = value
|
||
logger.debug(f"insert_by_path: inserted value '{value}' at path {path}")
|
||
except Exception as e:
|
||
logger.error(f"Error in insert_by_path: {e}, target: {target}, path: {path}, value: {value}")
|
||
|
||
|
||
def flatten_singleton_dict(obj):
|
||
"""
|
||
扁平化单元素字典，如果字典只有一个键且键名与值中的键名相同，则合并
|
||
|
||
Args:
|
||
obj: 待处理的对象
|
||
|
||
Returns:
|
||
处理后的对象
|
||
"""
|
||
try:
|
||
if not isinstance(obj, dict):
|
||
return obj
|
||
|
||
for k in list(obj.keys()):
|
||
v = obj[k]
|
||
if isinstance(v, dict):
|
||
v = flatten_singleton_dict(v)
|
||
if len(v) == 1 and k in v:
|
||
obj[k] = v[k]
|
||
logger.debug(f"flatten_singleton_dict: flattened singleton dict for key '{k}'")
|
||
else:
|
||
obj[k] = v
|
||
return obj
|
||
except Exception as e:
|
||
logger.error(f"Error in flatten_singleton_dict: {e}, obj: {obj}")
|
||
return obj
|
||
|
||
|
||
# def build_content(field_paths, row_content):
|
||
|
||
# try:
|
||
# result = {}
|
||
# for col, path in field_paths.items():
|
||
# value = normalize_value(row_content.get(col, ""))
|
||
# insert_by_path(result, path, value)
|
||
|
||
# flatten_singleton_dict(result)
|
||
# logger.debug(f"build_content: built content with {len(result)} top-level keys")
|
||
# return result
|
||
# except Exception as e:
|
||
# logger.error(f"Error in build_content: {e}, field_paths: {field_paths}, row_content: {row_content}")
|
||
# return {}
|
||
|
||
|
||
|
||
|
||
|
||
def flatten_dict(d, parent_key='', sep='_'):
|
||
"""
|
||
将 {"A": {"B": "1"}} 转换为 {"A_B": "1"}
|
||
"""
|
||
items = []
|
||
for k, v in d.items():
|
||
new_key = f"{parent_key}{sep}{k}" if parent_key else k
|
||
if isinstance(v, dict):
|
||
# 递归展开
|
||
items.extend(flatten_dict(v, new_key, sep=sep).items())
|
||
else:
|
||
items.append((new_key, v))
|
||
return dict(items)
|
||
|
||
def build_content(field_paths, row_content):
|
||
"""
|
||
根据字段路径构建内容结构
|
||
|
||
Args:
|
||
field_paths (dict): 字段路径映射
|
||
row_content (dict): 行内容数据
|
||
|
||
Returns:
|
||
dict: 构建的内容结构
|
||
"""
|
||
result = {}
|
||
|
||
for col, path in field_paths.items():
|
||
value = normalize_value(row_content.get(col, ""))
|
||
insert_by_path(result, path, value)
|
||
|
||
# 1. 先执行你原有的单级压缩 (解决 A -> {A: x})
|
||
flatten_singleton_dict(result)
|
||
|
||
# 2. 执行深度展平 (将嵌套字典转为 "键_子键" 格式)
|
||
# 我们遍历第一层，如果值是字典，则展开它
|
||
final_result = {}
|
||
for k, v in result.items():
|
||
if isinstance(v, dict):
|
||
# 展开子字典，并带上父级的键名作为前缀
|
||
flattened_sub = flatten_dict(v, parent_key=k, sep='_')
|
||
final_result.update(flattened_sub)
|
||
else:
|
||
final_result[k] = v
|
||
|
||
return final_result
|
||
|
||
|
||
# =====================================================
|
||
# 主解析逻辑
|
||
# =====================================================
|
||
|
||
def transform_excel_json(rows):
|
||
"""
|
||
将Excel行数据转换为JSON格式
|
||
|
||
Args:
|
||
rows (list): 行数据列表
|
||
|
||
Returns:
|
||
list: 转换后的JSON数据列表
|
||
"""
|
||
try:
|
||
logger.info(f"Starting transform_excel_json with {len(rows)} rows")
|
||
output = []
|
||
i = 0
|
||
processed_count = 0
|
||
|
||
while i < len(rows):
|
||
row = rows[i]
|
||
row_type = None
|
||
|
||
if is_header_row(row):
|
||
output.append({"header": row["content"]})
|
||
row_type = "header"
|
||
i += 1
|
||
elif is_field_row(row):
|
||
field_rows, next_idx = collect_field_rows(rows, i)
|
||
field_paths = build_field_paths(field_rows)
|
||
i = next_idx
|
||
row_type = "field"
|
||
|
||
while i < len(rows) and is_data_row(rows[i]):
|
||
content = build_content(field_paths, rows[i]["content"])
|
||
output.append({"content": content})
|
||
i += 1
|
||
processed_count += 1
|
||
else:
|
||
i += 1
|
||
continue # Skip non-header, non-field, non-data rows
|
||
|
||
logger.debug(f"Processed row {i-1}, type: {row_type}")
|
||
|
||
logger.info(f"Completed transform_excel_json: {len(output)} items, {processed_count} data rows processed")
|
||
return output
|
||
except Exception as e:
|
||
logger.error(f"Error in transform_excel_json: {e}")
|
||
logger.error(traceback.format_exc())
|
||
raise
|
||
|
||
|
||
# =====================================================
|
||
# Excel 读取（含合并单元格）
|
||
# =====================================================
|
||
|
||
def read_excel_with_merged_cells(file_path: str, sheet_name: Optional[str] = None):
|
||
"""
|
||
读取Excel文件，包括合并单元格的处理
|
||
|
||
Args:
|
||
file_path (str): Excel文件路径
|
||
sheet_name (Optional[str]): 工作表名称，如果为None则使用默认表
|
||
|
||
Returns:
|
||
pd.DataFrame: 包含Excel数据的DataFrame
|
||
"""
|
||
try:
|
||
_, ext = os.path.splitext(file_path)
|
||
ext = ext.lower()
|
||
if ext == ".xlsx":
|
||
logger.info(f"Reading Excel file: {file_path}, sheet: {sheet_name}")
|
||
wb = load_workbook(file_path, data_only=True)
|
||
ws = wb[sheet_name] if sheet_name else wb.active
|
||
|
||
max_row, max_col = ws.max_row, ws.max_column
|
||
logger.info(f"Excel sheet dimensions: {max_row} rows x {max_col} columns")
|
||
|
||
data = [["" for _ in range(max_col)] for _ in range(max_row)]
|
||
|
||
for r in range(1, max_row + 1):
|
||
for c in range(1, max_col + 1):
|
||
val = ws.cell(row=r, column=c).value
|
||
data[r - 1][c - 1] = "" if val is None else str(val).strip()
|
||
|
||
# 处理合并单元格
|
||
merged_count = 0
|
||
for merged in ws.merged_cells.ranges:
|
||
tl = data[merged.min_row - 1][merged.min_col - 1]
|
||
for r in range(merged.min_row - 1, merged.max_row):
|
||
for c in range(merged.min_col - 1, merged.max_col):
|
||
data[r][c] = tl
|
||
merged_count += 1
|
||
|
||
logger.info(f"Processed {merged_count} merged cell ranges")
|
||
df = pd.DataFrame(data)
|
||
logger.info(f"Created DataFrame with shape {df.shape}")
|
||
return df
|
||
|
||
else:
|
||
raise ValueError("Unsupported file format")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error in read_excel_with_merged_cells: {e}")
|
||
logger.error(traceback.format_exc())
|
||
raise
|
||
|
||
|
||
|
||
def dataframe_to_row_json(df: pd.DataFrame):
|
||
"""
|
||
将DataFrame转换为行JSON格式
|
||
|
||
Args:
|
||
df (pd.DataFrame): 源DataFrame
|
||
|
||
Returns:
|
||
list: 行JSON格式的列表
|
||
"""
|
||
try:
|
||
logger.info(f"Converting DataFrame to row JSON, shape: {df.shape}")
|
||
rows = []
|
||
for idx, row in df.iterrows():
|
||
content = {
|
||
str(col_idx + 1): val
|
||
for col_idx, val in enumerate(row)
|
||
if isinstance(val, str) and val != ""
|
||
}
|
||
if content:
|
||
rows.append({
|
||
"row_index": idx + 1,
|
||
"content": content
|
||
})
|
||
|
||
logger.info(f"Converted DataFrame to {len(rows)} JSON rows")
|
||
return rows
|
||
except Exception as e:
|
||
logger.error(f"Error in dataframe_to_row_json: {e}")
|
||
logger.error(traceback.format_exc())
|
||
raise
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# =====================================================
|
||
# Markdown 转换
|
||
# =====================================================
|
||
|
||
# def json_to_markdown(data):
|
||
# markdown_output = []
|
||
|
||
# for item in data:
|
||
# # 处理标题行 (Header)
|
||
# if "header" in item:
|
||
# # 提取 header 中的描述信息
|
||
# header_vals = item["header"].values()
|
||
# title = " ".join([v for v in header_vals if ";" in v or v.startswith("CPP") or "清单" in v])
|
||
# # 如果 header 中包含设备名称和型号，进行拼接
|
||
# if not title:
|
||
# title = " ".join(header_vals)
|
||
|
||
# markdown_output.append(f"\n## {title.replace(';', ' ')}\n")
|
||
|
||
# # 处理内容行 (Content)
|
||
# elif "content" in item:
|
||
# content = item["content"]
|
||
# # 如果是该表格的第一行，生成表头
|
||
# if not any("|" in line for line in markdown_output[-2:]):
|
||
# columns = list(content.keys())
|
||
# header_row = "| " + " | ".join(columns) + " |"
|
||
# separator_row = "| " + " | ".join(["---"] * len(columns)) + " |"
|
||
# markdown_output.append(header_row)
|
||
# markdown_output.append(separator_row)
|
||
|
||
# # 生成数据行
|
||
# values = [str(v) if v is not None else "" for v in content.values()]
|
||
# # 处理布尔值显示
|
||
# values = ["是" if v == "True" else "否" if v == "False" else v for v in values]
|
||
# markdown_output.append("| " + " | ".join(values) + " |")
|
||
|
||
# return "\n".join(markdown_output)
|
||
|
||
|
||
|
||
|
||
|
||
import json
|
||
|
||
def json_to_markdown(json_data):
|
||
markdown_lines = []
|
||
current_table_data = []
|
||
|
||
for entry in json_data:
|
||
# 1. 处理标题行 (Header)
|
||
if "header" in entry:
|
||
# 如果之前已经有累积的数据，先输出旧表格（防止连续多个表格的情况）
|
||
if current_table_data:
|
||
markdown_lines.extend(build_table(current_table_data))
|
||
current_table_data = []
|
||
|
||
# 提取 Header 内容并格式化为标题
|
||
header_dict = entry["header"]
|
||
# 按照您的要求，处理设备名称和型号的拼接显示
|
||
title_parts = [v.replace(';', ' ') for v in header_dict.values()]
|
||
title = " ".join(title_parts)
|
||
markdown_lines.append(f"\n## {title}\n")
|
||
|
||
# 2. 处理内容行 (Content)
|
||
elif "content" in entry:
|
||
current_table_data.append(entry["content"])
|
||
|
||
# 3. 输出最后一个表格
|
||
if current_table_data:
|
||
markdown_lines.extend(build_table(current_table_data))
|
||
|
||
return "\n".join(markdown_lines)
|
||
|
||
def build_table(data_list):
|
||
"""将字典列表转换为 Markdown 表格字符串"""
|
||
if not data_list:
|
||
return []
|
||
|
||
lines = []
|
||
# 提取所有键作为表头
|
||
headers = list(data_list[0].keys())
|
||
|
||
# 构造表头行
|
||
header_line = "| " + " | ".join(headers) + " |"
|
||
# 构造分割行
|
||
separator_line = "| " + " | ".join(["---"] * len(headers)) + " |"
|
||
|
||
lines.append(header_line)
|
||
lines.append(separator_line)
|
||
|
||
# 构造数据行
|
||
for item in data_list:
|
||
row_values = []
|
||
for h in headers:
|
||
val = item.get(h, "")
|
||
# 转换布尔值为中文，处理 None 值
|
||
if val is True: val = "是"
|
||
elif val is False or val == "": val = " "
|
||
row_values.append(str(val).replace("\n", " ")) # 避免数据内换行破坏表格结构
|
||
|
||
lines.append("| " + " | ".join(row_values) + " |")
|
||
|
||
return lines
|
||
|
||
# ---------------------------------------------------------
|
||
# 执行转换
|
||
# ---------------------------------------------------------
|
||
# 假设 source_json 是您提供的那段原始数据
|
||
try:
|
||
# 这里的 source_json 代表您贴出的完整 JSON 列表
|
||
# md_output = json_to_markdown_tables(source_json)
|
||
# print(md_output)
|
||
pass
|
||
except Exception as e:
|
||
print(f"解析错误: {e}")
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# def flatten_dict(d, parent_key='', sep='（', end_sep='）'):
|
||
# """
|
||
# 递归扁平化嵌套字典，生成 '父字段（子字段）（孙字段）' 格式的键。
|
||
# 支持任意深度嵌套。
|
||
# """
|
||
# items = []
|
||
# for k, v in d.items():
|
||
# new_key = f"{parent_key}{sep}{k}{end_sep}" if parent_key else k
|
||
# if isinstance(v, dict):
|
||
# items.extend(flatten_dict(v, new_key, sep, end_sep).items())
|
||
# else:
|
||
# items.append((new_key, v))
|
||
# return dict(items)
|
||
|
||
# def extract_all_content_records(data):
|
||
# """
|
||
# 从任意结构的 JSON 数据中提取所有顶层包含 "content" 键的对象。
|
||
# 假设数据是一个列表，每个元素可能是 header 或 content。
|
||
# 如果不是列表，也尝试遍历。
|
||
# """
|
||
# records = []
|
||
|
||
# def recurse(obj):
|
||
# if isinstance(obj, list):
|
||
# for item in obj:
|
||
# recurse(item)
|
||
# elif isinstance(obj, dict):
|
||
# if "content" in obj and isinstance(obj["content"], dict):
|
||
# records.append(obj["content"])
|
||
# else:
|
||
# for value in obj.values():
|
||
# recurse(value)
|
||
|
||
# recurse(data)
|
||
# return records
|
||
|
||
# def build_markdown_table(records):
|
||
# if not records:
|
||
# return "> 未找到任何 content 记录。\n"
|
||
|
||
# flat_records = [flatten_dict(record) for record in records]
|
||
|
||
# # 收集所有字段（保持顺序）
|
||
# all_keys = {}
|
||
# for fr in flat_records:
|
||
# for k in fr:
|
||
# all_keys[k] = None
|
||
# all_keys = list(all_keys.keys())
|
||
|
||
# # 构建表格
|
||
# header = "| " + " | ".join(all_keys) + " |"
|
||
# separator = "| " + " | ".join(["---"] * len(all_keys)) + " |"
|
||
# rows = [
|
||
# "| " + " | ".join(str(fr.get(k, "")) for k in all_keys) + " |"
|
||
# for fr in flat_records
|
||
# ]
|
||
|
||
# return "\n".join([header, separator] + rows)
|
||
|
||
# def json_to_markdown(json_data):
|
||
# """
|
||
# 将任意 JSON 文件中所有 'content' 对象转换为 Markdown 表格。
|
||
# """
|
||
|
||
|
||
# records = extract_all_content_records(json_data)
|
||
# md_table = build_markdown_table(records)
|
||
# return md_table
|
||
|
||
|
||
|
||
|