qx-1-codingshutong-task · hzy9588/数通设备自动化运维-python - AtomGitimport re def main(text: str) - dict: Dify Code节点解析Markdown表格标准化列名并按预定义表结构重排列顺序。 仅使用Python标准库兼容Dify沙箱环境。 # 1. 配置定义 # 列名映射表key统一小写用于模糊匹配原始表头 COLUMN_MAPPING { 客户id: customer_id, 客户编号: customer_id, 投诉id: complaint_id, 员工姓名: handler_name, 员工id: handler_id, handler_id: handler_id, 营业厅id: hall_id, hall_id: hall_id, 营业厅名称: hall_name, 区域: region, 接单时间: accept_time, 套餐类型: package_type, 投诉内容: complaint_description, 投诉类型: complaint_type, 投诉子类: complaint_subtype, 客户等级: customer_level, } # 表结构定义表名 - 固定列顺序列表 TABLE_SCHEMAS { complaint_record: [ complaint_id, accept_time, customer_id, handler_name, package_type, complaint_description, complaint_type, complaint_subtype ], complaint_handling: [ complaint_id, handler_id, hall_id, accept_time, customer_feedback, customer_name ], customer_info: [ customer_id, customer_name, customer_level ], employee_info: [ handler_id, handler_name, hall_id ], hall_info: [ hall_id, hall_name, region ], } # 2. 辅助函数 def normalize_header(raw_header: str) - str: 将原始表头转为标准字段名未匹配则原样保留去空格后的小写形式 cleaned raw_header.strip() key cleaned.lower().replace(_, ).replace(-, ).replace( , ) # 尝试直接匹配先去分隔符再匹配 for map_key, std_name in COLUMN_MAPPING.items(): map_key_norm map_key.lower().replace(_, ).replace(-, ).replace( , ) if key map_key_norm: return std_name # 未匹配到映射表返回去除首尾空格的原始值 return cleaned def identify_table_type(headers: list) - str: 根据标准化后的表头与预定义schema的重合度智能识别表类型。 返回最佳匹配的表名若无匹配返回None。 header_set set(headers) best_match None best_score 0 for table_name, schema_cols in TABLE_SCHEMAS.items(): schema_set set(schema_cols) # 计算交集大小作为重合度得分 intersection header_set schema_set score len(intersection) # 只有当交集至少覆盖schema一半字段时才视为有效匹配 if score best_score and score max(2, len(schema_set) // 2): best_score score best_match table_name return best_match def reorder_columns(headers: list, rows: list, table_type: str) - tuple: 按预定义schema重排列顺序额外列追加到末尾。 返回 (新headers, 新rows)。 if table_type is None or table_type not in TABLE_SCHEMAS: return headers, rows schema_order TABLE_SCHEMAS[table_type] # 构建新列顺序先按schema顺序再追加未在schema中的额外列 extra_cols [h for h in headers if h not in schema_order] new_order [col for col in schema_order if col in headers] extra_cols # 建立旧索引到新索引的映射 old_index_map {h: i for i, h in enumerate(headers)} new_indices [] for col in new_order: if col in old_index_map: new_indices.append(old_index_map[col]) # 重排每一行数据 new_rows [] for row in rows: new_row [] for idx in new_indices: if idx len(row): new_row.append(row[idx]) else: new_row.append() # 列数不足时补空 new_rows.append(new_row) return new_order, new_rows def parse_markdown_table(table_text: str) - dict: 解析单个Markdown表格文本返回 {headers, rows, raw_lines}。 若解析失败返回None。 lines [l for l in table_text.strip().split(\n) if l.strip()] if len(lines) 3: return None # 验证是否为合法Markdown表格需含分隔行 |---| sep_found False sep_idx -1 for i, line in enumerate(lines): stripped line.strip() if re.match(r^\|[\s\-:|]\|$, stripped): sep_found True sep_idx i break if not sep_found or sep_idx 0: return None # 解析表头分隔行之前的所有行合并为表头通常只有一行 header_line lines[sep_idx - 1].strip() raw_headers [c.strip() for c in header_line.strip(|).split(|)] # 解析数据行分隔行之后的所有行 data_lines lines[sep_idx 1:] rows [] for dl in data_lines: stripped dl.strip() if not stripped.startswith(|): continue cells [c.strip() for c in stripped.strip(|).split(|)] rows.append(cells) return {raw_headers: raw_headers, rows: rows} def extract_tables(full_text: str) - list: 从全文中提取所有Markdown表格块连续含|的行组。 返回表格原文字符串列表。 tables [] current_block [] in_table False for line in full_text.split(\n): stripped line.strip() is_table_line | in stripped and stripped.startswith(|) if is_table_line: current_block.append(line) in_table True else: if in_table and current_block: tables.append(\n.join(current_block)) current_block [] in_table False # 处理末尾可能残留的表格块 if current_block: tables.append(\n.join(current_block)) return tables def rebuild_markdown_table(headers: list, rows: list) - str: 将headers和rows重新组装为Markdown表格字符串 if not headers: return header_line | | .join(headers) | sep_line | | .join([---] * len(headers)) | data_lines [] for row in rows: # 确保每行列数与header一致 padded row [] * (len(headers) - len(row)) data_lines.append(| | .join(padded[:len(headers)]) |) return \n.join([header_line, sep_line] data_lines) # 3. 主处理流程 try: table_blocks extract_tables(text) if not table_blocks: # 未找到任何表格原样返回 return {table: text} result_parts [] for block in table_blocks: parsed parse_markdown_table(block) if parsed is None: # 无法解析的表格块原样保留 result_parts.append(block) continue raw_headers parsed[raw_headers] rows parsed[rows] # Step1: 列名标准化 std_headers [normalize_header(h) for h in raw_headers] # Step2: 识别表类型 table_type identify_table_type(std_headers) # Step3: 按schema重排列顺序数据行内容不变 final_headers, final_rows reorder_columns(std_headers, rows, table_type) # Step4: 重建Markdown表格 rebuilt rebuild_markdown_table(final_headers, final_rows) result_parts.append(rebuilt) result_text \n\n.join(result_parts) return {table: result_text} except Exception: # 任何异常均不抛出返回原始文本保证节点不崩溃 return {table: text}qx-2-codingfrom datetime import datetime from typing import Dict def main(text: str) - Dict[str, str]: 输入包含一个或多个 Markdown 表格的文本字符串 输出字典 {table: 清洗后的 Markdown 表格字符串} # 定义所有可识别的时间格式按优先级顺序 TIME_FORMATS [ %m/%d/%Y %I:%M %p, # 美式12小时带AM/PM %m/%d/%Y %H:%M, # 美式24小时 %Y/%m/%d %H:%M, # 斜杠格式 %Y.%m.%d %H:%M, # 点分格式 %d-%m-%Y %H:%M, # 欧式 %b %d, %Y %H:%M, # 英文月份简写 %d %B %Y %H:%M, # 全英文月份 %A, %d %B %Y %I:%M %p, # 带星期英文 %Y年%m月%d日 %H:%M, # 中文年月日 %Y%m%d%H%M%S, # 紧凑数字需14位 %H:%M %d-%m-%Y, # 时间在前 ] def convert_time(value: str) - str: 尝试将时间字符串转换为 %Y/%m/%d %H:%M失败则返回原值 if not value or not value.strip(): return value s value.strip() for fmt in TIME_FORMATS: try: dt datetime.strptime(s, fmt) return dt.strftime(%Y/%m/%d %H:%M) except ValueError: continue return value # 无法识别保留原样 def split_cells(line: str) - list: 解析 Markdown 表格的一行返回单元格列表已去除首尾空格 return [c.strip() for c in line.strip(|).split(|)] def process_table(rows: list) - list: 处理单个表格的行列表包含表头、分隔行、数据行。 返回处理后的新行列表。 if len(rows) 3: return rows # 无效表格原样返回 header rows[0] sep rows[1] data_rows rows[2:] header_cells split_cells(header) # 查找 accept_time 列索引忽略大小写 idx None for i, name in enumerate(header_cells): if name.lower() accept_time: idx i break if idx is None: # 未找到目标列整表原样返回 return rows # 处理数据行 new_data [] for row in data_rows: cells split_cells(row) if idx len(cells): cells[idx] convert_time(cells[idx]) # 重新拼接为 Markdown 行 new_data.append(| | .join(cells) |) # 重新生成表头行 new_header | | .join(header_cells) | # 重新生成分隔行保留列数 col_count len(header_cells) new_sep | | .join([---] * col_count) | return [new_header, new_sep] new_data # 1. 按行拆分输入文本 lines text.splitlines() # 2. 提取所有完整的 Markdown 表格连续以 | 开头和结尾的行 tables [] i 0 while i len(lines): line lines[i].strip() if line.startswith(|) and line.endswith(|): rows [] while i len(lines) and lines[i].strip().startswith(|) and lines[i].strip().endswith(|): rows.append(lines[i].strip()) i 1 if len(rows) 3: # 至少表头、分隔、一行数据 tables.append(rows) else: i 1 # 3. 依次处理每个表格 processed_tables [] for tbl in tables: processed_tables.append(process_table(tbl)) # 4. 将所有表格重新拼接为一个 Markdown 字符串表格间空行分隔 output_lines [] for tbl in processed_tables: output_lines.extend(tbl) output_lines.append() # 空行分隔 # 移除末尾多余空行 while output_lines and output_lines[-1] : output_lines.pop() result_md \n.join(output_lines) return {table: result_md}qx-2-biangengimport re from datetime import datetime def parse_markdown_table(markdown_str): 解析Markdown格式的表格返回字段名列表和数据行列表 正确保留空值 if not markdown_str or not markdown_str.strip(): return [], [] lines markdown_str.strip().split(\n) if len(lines) 3: return [], [] # 解析表头 header_line lines[0].strip() header_parts header_line.split(|) # 去除首尾的空字符串 if header_parts and header_parts[0] : header_parts header_parts[1:] if header_parts and header_parts[-1] : header_parts header_parts[:-1] headers [h.strip() for h in header_parts] # 跳过分隔符行 data_lines lines[2:] data_rows [] for line in data_lines: if | in line: parts line.split(|) # 去除首尾的空字符串 if parts and parts[0] : parts parts[1:] if parts and parts[-1] : parts parts[:-1] # 保留空值 cells [p.strip() for p in parts] # 确保单元格数量与表头一致 if len(cells) len(headers): cells cells [] * (len(headers) - len(cells)) elif len(cells) len(headers): cells cells[:len(headers)] data_rows.append(cells) return headers, data_rows def is_datetime_format(value): 检查值是否符合任一日期时间格式 返回True表示符合某种格式False表示不符合 if not value or not value.strip(): return False value value.strip() # 定义所有支持的格式模式 patterns [ # 点分格式 %Y.%m.%d %H:%M r^\d{4}\.\d{1,2}\.\d{1,2} \d{1,2}:\d{2}$, # 欧式 %d-%m-%Y %H:%M r^\d{1,2}-\d{1,2}-\d{4} \d{1,2}:\d{2}$, # 美式12小时 %m/%d/%Y %I:%M %p r^\d{1,2}/\d{1,2}/\d{4} \d{1,2}:\d{2} (AM|PM)$, # 英文月份 %b %d, %Y %H:%M r^[A-Za-z]{3} \d{1,2}, \d{4} \d{1,2}:\d{2}$, # 全英文月份 %d %B %Y %H:%M r^\d{1,2} [A-Za-z] \d{4} \d{1,2}:\d{2}$, # 带星期英文 %A, %d %B %Y %I:%M %p r^[A-Za-z], \d{1,2} [A-Za-z] \d{4} \d{1,2}:\d{2} (AM|PM)$, # 中文年月日 %Y年%m月%d日 %H:%M r^\d{4}年\d{1,2}月\d{1,2}日 \d{1,2}:\d{2}$, # 紧凑数字 %Y%m%d%H%M%S r^\d{14}$, # 仅日期 %Y/%m/%d r^\d{4}/\d{1,2}/\d{1,2}$, # 时间在前 %H:%M %d-%m-%Y r^\d{1,2}:\d{2} \d{1,2}-\d{1,2}-\d{4}$, # 标准格式补充 r^\d{4}/\d{1,2}/\d{1,2} \d{1,2}:\d{2}$, r^\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{2}$, ] for pattern in patterns: if re.match(pattern, value, re.IGNORECASE): return True return False def count_fixed_rows(old_data, new_data, old_accept_idx, new_accept_idx): 统计清洗修复行数 逐行比较清洗前accept_time如果满足日期时间格式则算一次修复 if not old_data or not new_data: return 0 fixed_count 0 min_rows min(len(old_data), len(new_data)) for i in range(min_rows): old_time old_data[i][old_accept_idx] if old_accept_idx len(old_data[i]) else # 如果旧值符合任一日期时间格式算一次修复 if is_datetime_format(old_time): fixed_count 1 return fixed_count def generate_markdown_table(headers, data_rows, max_rows6): 生成Markdown格式表格保留空值 if not headers: return 无数据 # 取前max_rows行 display_rows data_rows[:max_rows] if data_rows else [] # 构建表头 header_line | | .join(headers) | separator_line | | .join([---] * len(headers)) | if not display_rows: return \n.join([header_line, separator_line]) # 构建数据行 data_lines [] for row in display_rows: # 确保列数一致 if len(row) len(headers): row row [] * (len(headers) - len(row)) elif len(row) len(headers): row row[:len(headers)] data_lines.append(| | .join(row) |) return \n.join([header_line, separator_line] data_lines) def main(o, n, o1, n1): 主函数生成数据清洗变更报告 参数: o: complaint_record清洗前的Markdown表格字符串 n: complaint_record清洗后的Markdown表格字符串 o1: complaint_handling清洗前的Markdown表格字符串 n1: complaint_handling清洗后的Markdown表格字符串 返回: 字典: {baogao: 数据清洗变更报告内容} report_parts [# 数据清洗变更报告] # 处理 complaint_record 表 report_parts.append(## 表complaint_record) # 解析清洗前后的数据 headers_r_old, old_data_r parse_markdown_table(o) headers_r_new, new_data_r parse_markdown_table(n) # 原始数据行数 original_count_r len(old_data_r) report_parts.append(f- 原始数据行数{original_count_r}) # 清洗修复行数 if headers_r_old and headers_r_new and old_data_r and new_data_r: # 找到 accept_time 字段索引 old_accept_idx -1 new_accept_idx -1 for idx, h in enumerate(headers_r_old): if accept_time in h.lower(): old_accept_idx idx break for idx, h in enumerate(headers_r_new): if accept_time in h.lower(): new_accept_idx idx break if old_accept_idx ! -1 and new_accept_idx ! -1: fixed_count count_fixed_rows( old_data_r, new_data_r, old_accept_idx, new_accept_idx ) report_parts.append(f- 清洗修复行数{fixed_count}) else: report_parts.append(- 清洗修复行数0未找到accept_time字段) # 结果预览前6行 report_parts.append(- 结果预览前6行) table_preview generate_markdown_table(headers_r_new, new_data_r, 6) report_parts.append(table_preview) else: report_parts.append(- 清洗修复行数0) report_parts.append(- 结果预览前6行无数据) # 处理 complaint_handling 表 report_parts.append(## 表complaint_handling) # 解析清洗前后的数据 headers_h_old, old_data_h parse_markdown_table(o1) headers_h_new, new_data_h parse_markdown_table(n1) # 原始数据行数 original_count_h len(old_data_h) report_parts.append(f- 原始数据行数{original_count_h}) # 清洗修复行数 if headers_h_old and headers_h_new and old_data_h and new_data_h: # 找到 accept_time 字段索引 old_accept_idx -1 new_accept_idx -1 for idx, h in enumerate(headers_h_old): if accept_time in h.lower(): old_accept_idx idx break for idx, h in enumerate(headers_h_new): if accept_time in h.lower(): new_accept_idx idx break if old_accept_idx ! -1 and new_accept_idx ! -1: fixed_count count_fixed_rows( old_data_h, new_data_h, old_accept_idx, new_accept_idx ) report_parts.append(f- 清洗修复行数{fixed_count}) else: report_parts.append(- 清洗修复行数0未找到accept_time字段) # 结果预览前6行 report_parts.append(- 结果预览前6行) table_preview generate_markdown_table(headers_h_new, new_data_h, 6) report_parts.append(table_preview) else: report_parts.append(- 清洗修复行数0) report_parts.append(- 结果预览前6行无数据) # 组合报告 report_content \n\n.join(report_parts) return {baogao: report_content}
qx-1-codingshutong-task · hzy9588/数通设备自动化运维-python - AtomGitimport re def main(text: str) - dict: Dify Code节点解析Markdown表格标准化列名并按预定义表结构重排列顺序。 仅使用Python标准库兼容Dify沙箱环境。 # 1. 配置定义 # 列名映射表key统一小写用于模糊匹配原始表头 COLUMN_MAPPING { 客户id: customer_id, 客户编号: customer_id, 投诉id: complaint_id, 员工姓名: handler_name, 员工id: handler_id, handler_id: handler_id, 营业厅id: hall_id, hall_id: hall_id, 营业厅名称: hall_name, 区域: region, 接单时间: accept_time, 套餐类型: package_type, 投诉内容: complaint_description, 投诉类型: complaint_type, 投诉子类: complaint_subtype, 客户等级: customer_level, } # 表结构定义表名 - 固定列顺序列表 TABLE_SCHEMAS { complaint_record: [ complaint_id, accept_time, customer_id, handler_name, package_type, complaint_description, complaint_type, complaint_subtype ], complaint_handling: [ complaint_id, handler_id, hall_id, accept_time, customer_feedback, customer_name ], customer_info: [ customer_id, customer_name, customer_level ], employee_info: [ handler_id, handler_name, hall_id ], hall_info: [ hall_id, hall_name, region ], } # 2. 辅助函数 def normalize_header(raw_header: str) - str: 将原始表头转为标准字段名未匹配则原样保留去空格后的小写形式 cleaned raw_header.strip() key cleaned.lower().replace(_, ).replace(-, ).replace( , ) # 尝试直接匹配先去分隔符再匹配 for map_key, std_name in COLUMN_MAPPING.items(): map_key_norm map_key.lower().replace(_, ).replace(-, ).replace( , ) if key map_key_norm: return std_name # 未匹配到映射表返回去除首尾空格的原始值 return cleaned def identify_table_type(headers: list) - str: 根据标准化后的表头与预定义schema的重合度智能识别表类型。 返回最佳匹配的表名若无匹配返回None。 header_set set(headers) best_match None best_score 0 for table_name, schema_cols in TABLE_SCHEMAS.items(): schema_set set(schema_cols) # 计算交集大小作为重合度得分 intersection header_set schema_set score len(intersection) # 只有当交集至少覆盖schema一半字段时才视为有效匹配 if score best_score and score max(2, len(schema_set) // 2): best_score score best_match table_name return best_match def reorder_columns(headers: list, rows: list, table_type: str) - tuple: 按预定义schema重排列顺序额外列追加到末尾。 返回 (新headers, 新rows)。 if table_type is None or table_type not in TABLE_SCHEMAS: return headers, rows schema_order TABLE_SCHEMAS[table_type] # 构建新列顺序先按schema顺序再追加未在schema中的额外列 extra_cols [h for h in headers if h not in schema_order] new_order [col for col in schema_order if col in headers] extra_cols # 建立旧索引到新索引的映射 old_index_map {h: i for i, h in enumerate(headers)} new_indices [] for col in new_order: if col in old_index_map: new_indices.append(old_index_map[col]) # 重排每一行数据 new_rows [] for row in rows: new_row [] for idx in new_indices: if idx len(row): new_row.append(row[idx]) else: new_row.append() # 列数不足时补空 new_rows.append(new_row) return new_order, new_rows def parse_markdown_table(table_text: str) - dict: 解析单个Markdown表格文本返回 {headers, rows, raw_lines}。 若解析失败返回None。 lines [l for l in table_text.strip().split(\n) if l.strip()] if len(lines) 3: return None # 验证是否为合法Markdown表格需含分隔行 |---| sep_found False sep_idx -1 for i, line in enumerate(lines): stripped line.strip() if re.match(r^\|[\s\-:|]\|$, stripped): sep_found True sep_idx i break if not sep_found or sep_idx 0: return None # 解析表头分隔行之前的所有行合并为表头通常只有一行 header_line lines[sep_idx - 1].strip() raw_headers [c.strip() for c in header_line.strip(|).split(|)] # 解析数据行分隔行之后的所有行 data_lines lines[sep_idx 1:] rows [] for dl in data_lines: stripped dl.strip() if not stripped.startswith(|): continue cells [c.strip() for c in stripped.strip(|).split(|)] rows.append(cells) return {raw_headers: raw_headers, rows: rows} def extract_tables(full_text: str) - list: 从全文中提取所有Markdown表格块连续含|的行组。 返回表格原文字符串列表。 tables [] current_block [] in_table False for line in full_text.split(\n): stripped line.strip() is_table_line | in stripped and stripped.startswith(|) if is_table_line: current_block.append(line) in_table True else: if in_table and current_block: tables.append(\n.join(current_block)) current_block [] in_table False # 处理末尾可能残留的表格块 if current_block: tables.append(\n.join(current_block)) return tables def rebuild_markdown_table(headers: list, rows: list) - str: 将headers和rows重新组装为Markdown表格字符串 if not headers: return header_line | | .join(headers) | sep_line | | .join([---] * len(headers)) | data_lines [] for row in rows: # 确保每行列数与header一致 padded row [] * (len(headers) - len(row)) data_lines.append(| | .join(padded[:len(headers)]) |) return \n.join([header_line, sep_line] data_lines) # 3. 主处理流程 try: table_blocks extract_tables(text) if not table_blocks: # 未找到任何表格原样返回 return {table: text} result_parts [] for block in table_blocks: parsed parse_markdown_table(block) if parsed is None: # 无法解析的表格块原样保留 result_parts.append(block) continue raw_headers parsed[raw_headers] rows parsed[rows] # Step1: 列名标准化 std_headers [normalize_header(h) for h in raw_headers] # Step2: 识别表类型 table_type identify_table_type(std_headers) # Step3: 按schema重排列顺序数据行内容不变 final_headers, final_rows reorder_columns(std_headers, rows, table_type) # Step4: 重建Markdown表格 rebuilt rebuild_markdown_table(final_headers, final_rows) result_parts.append(rebuilt) result_text \n\n.join(result_parts) return {table: result_text} except Exception: # 任何异常均不抛出返回原始文本保证节点不崩溃 return {table: text}qx-2-codingfrom datetime import datetime from typing import Dict def main(text: str) - Dict[str, str]: 输入包含一个或多个 Markdown 表格的文本字符串 输出字典 {table: 清洗后的 Markdown 表格字符串} # 定义所有可识别的时间格式按优先级顺序 TIME_FORMATS [ %m/%d/%Y %I:%M %p, # 美式12小时带AM/PM %m/%d/%Y %H:%M, # 美式24小时 %Y/%m/%d %H:%M, # 斜杠格式 %Y.%m.%d %H:%M, # 点分格式 %d-%m-%Y %H:%M, # 欧式 %b %d, %Y %H:%M, # 英文月份简写 %d %B %Y %H:%M, # 全英文月份 %A, %d %B %Y %I:%M %p, # 带星期英文 %Y年%m月%d日 %H:%M, # 中文年月日 %Y%m%d%H%M%S, # 紧凑数字需14位 %H:%M %d-%m-%Y, # 时间在前 ] def convert_time(value: str) - str: 尝试将时间字符串转换为 %Y/%m/%d %H:%M失败则返回原值 if not value or not value.strip(): return value s value.strip() for fmt in TIME_FORMATS: try: dt datetime.strptime(s, fmt) return dt.strftime(%Y/%m/%d %H:%M) except ValueError: continue return value # 无法识别保留原样 def split_cells(line: str) - list: 解析 Markdown 表格的一行返回单元格列表已去除首尾空格 return [c.strip() for c in line.strip(|).split(|)] def process_table(rows: list) - list: 处理单个表格的行列表包含表头、分隔行、数据行。 返回处理后的新行列表。 if len(rows) 3: return rows # 无效表格原样返回 header rows[0] sep rows[1] data_rows rows[2:] header_cells split_cells(header) # 查找 accept_time 列索引忽略大小写 idx None for i, name in enumerate(header_cells): if name.lower() accept_time: idx i break if idx is None: # 未找到目标列整表原样返回 return rows # 处理数据行 new_data [] for row in data_rows: cells split_cells(row) if idx len(cells): cells[idx] convert_time(cells[idx]) # 重新拼接为 Markdown 行 new_data.append(| | .join(cells) |) # 重新生成表头行 new_header | | .join(header_cells) | # 重新生成分隔行保留列数 col_count len(header_cells) new_sep | | .join([---] * col_count) | return [new_header, new_sep] new_data # 1. 按行拆分输入文本 lines text.splitlines() # 2. 提取所有完整的 Markdown 表格连续以 | 开头和结尾的行 tables [] i 0 while i len(lines): line lines[i].strip() if line.startswith(|) and line.endswith(|): rows [] while i len(lines) and lines[i].strip().startswith(|) and lines[i].strip().endswith(|): rows.append(lines[i].strip()) i 1 if len(rows) 3: # 至少表头、分隔、一行数据 tables.append(rows) else: i 1 # 3. 依次处理每个表格 processed_tables [] for tbl in tables: processed_tables.append(process_table(tbl)) # 4. 将所有表格重新拼接为一个 Markdown 字符串表格间空行分隔 output_lines [] for tbl in processed_tables: output_lines.extend(tbl) output_lines.append() # 空行分隔 # 移除末尾多余空行 while output_lines and output_lines[-1] : output_lines.pop() result_md \n.join(output_lines) return {table: result_md}qx-2-biangengimport re from datetime import datetime def parse_markdown_table(markdown_str): 解析Markdown格式的表格返回字段名列表和数据行列表 正确保留空值 if not markdown_str or not markdown_str.strip(): return [], [] lines markdown_str.strip().split(\n) if len(lines) 3: return [], [] # 解析表头 header_line lines[0].strip() header_parts header_line.split(|) # 去除首尾的空字符串 if header_parts and header_parts[0] : header_parts header_parts[1:] if header_parts and header_parts[-1] : header_parts header_parts[:-1] headers [h.strip() for h in header_parts] # 跳过分隔符行 data_lines lines[2:] data_rows [] for line in data_lines: if | in line: parts line.split(|) # 去除首尾的空字符串 if parts and parts[0] : parts parts[1:] if parts and parts[-1] : parts parts[:-1] # 保留空值 cells [p.strip() for p in parts] # 确保单元格数量与表头一致 if len(cells) len(headers): cells cells [] * (len(headers) - len(cells)) elif len(cells) len(headers): cells cells[:len(headers)] data_rows.append(cells) return headers, data_rows def is_datetime_format(value): 检查值是否符合任一日期时间格式 返回True表示符合某种格式False表示不符合 if not value or not value.strip(): return False value value.strip() # 定义所有支持的格式模式 patterns [ # 点分格式 %Y.%m.%d %H:%M r^\d{4}\.\d{1,2}\.\d{1,2} \d{1,2}:\d{2}$, # 欧式 %d-%m-%Y %H:%M r^\d{1,2}-\d{1,2}-\d{4} \d{1,2}:\d{2}$, # 美式12小时 %m/%d/%Y %I:%M %p r^\d{1,2}/\d{1,2}/\d{4} \d{1,2}:\d{2} (AM|PM)$, # 英文月份 %b %d, %Y %H:%M r^[A-Za-z]{3} \d{1,2}, \d{4} \d{1,2}:\d{2}$, # 全英文月份 %d %B %Y %H:%M r^\d{1,2} [A-Za-z] \d{4} \d{1,2}:\d{2}$, # 带星期英文 %A, %d %B %Y %I:%M %p r^[A-Za-z], \d{1,2} [A-Za-z] \d{4} \d{1,2}:\d{2} (AM|PM)$, # 中文年月日 %Y年%m月%d日 %H:%M r^\d{4}年\d{1,2}月\d{1,2}日 \d{1,2}:\d{2}$, # 紧凑数字 %Y%m%d%H%M%S r^\d{14}$, # 仅日期 %Y/%m/%d r^\d{4}/\d{1,2}/\d{1,2}$, # 时间在前 %H:%M %d-%m-%Y r^\d{1,2}:\d{2} \d{1,2}-\d{1,2}-\d{4}$, # 标准格式补充 r^\d{4}/\d{1,2}/\d{1,2} \d{1,2}:\d{2}$, r^\d{4}-\d{1,2}-\d{1,2} \d{1,2}:\d{2}$, ] for pattern in patterns: if re.match(pattern, value, re.IGNORECASE): return True return False def count_fixed_rows(old_data, new_data, old_accept_idx, new_accept_idx): 统计清洗修复行数 逐行比较清洗前accept_time如果满足日期时间格式则算一次修复 if not old_data or not new_data: return 0 fixed_count 0 min_rows min(len(old_data), len(new_data)) for i in range(min_rows): old_time old_data[i][old_accept_idx] if old_accept_idx len(old_data[i]) else # 如果旧值符合任一日期时间格式算一次修复 if is_datetime_format(old_time): fixed_count 1 return fixed_count def generate_markdown_table(headers, data_rows, max_rows6): 生成Markdown格式表格保留空值 if not headers: return 无数据 # 取前max_rows行 display_rows data_rows[:max_rows] if data_rows else [] # 构建表头 header_line | | .join(headers) | separator_line | | .join([---] * len(headers)) | if not display_rows: return \n.join([header_line, separator_line]) # 构建数据行 data_lines [] for row in display_rows: # 确保列数一致 if len(row) len(headers): row row [] * (len(headers) - len(row)) elif len(row) len(headers): row row[:len(headers)] data_lines.append(| | .join(row) |) return \n.join([header_line, separator_line] data_lines) def main(o, n, o1, n1): 主函数生成数据清洗变更报告 参数: o: complaint_record清洗前的Markdown表格字符串 n: complaint_record清洗后的Markdown表格字符串 o1: complaint_handling清洗前的Markdown表格字符串 n1: complaint_handling清洗后的Markdown表格字符串 返回: 字典: {baogao: 数据清洗变更报告内容} report_parts [# 数据清洗变更报告] # 处理 complaint_record 表 report_parts.append(## 表complaint_record) # 解析清洗前后的数据 headers_r_old, old_data_r parse_markdown_table(o) headers_r_new, new_data_r parse_markdown_table(n) # 原始数据行数 original_count_r len(old_data_r) report_parts.append(f- 原始数据行数{original_count_r}) # 清洗修复行数 if headers_r_old and headers_r_new and old_data_r and new_data_r: # 找到 accept_time 字段索引 old_accept_idx -1 new_accept_idx -1 for idx, h in enumerate(headers_r_old): if accept_time in h.lower(): old_accept_idx idx break for idx, h in enumerate(headers_r_new): if accept_time in h.lower(): new_accept_idx idx break if old_accept_idx ! -1 and new_accept_idx ! -1: fixed_count count_fixed_rows( old_data_r, new_data_r, old_accept_idx, new_accept_idx ) report_parts.append(f- 清洗修复行数{fixed_count}) else: report_parts.append(- 清洗修复行数0未找到accept_time字段) # 结果预览前6行 report_parts.append(- 结果预览前6行) table_preview generate_markdown_table(headers_r_new, new_data_r, 6) report_parts.append(table_preview) else: report_parts.append(- 清洗修复行数0) report_parts.append(- 结果预览前6行无数据) # 处理 complaint_handling 表 report_parts.append(## 表complaint_handling) # 解析清洗前后的数据 headers_h_old, old_data_h parse_markdown_table(o1) headers_h_new, new_data_h parse_markdown_table(n1) # 原始数据行数 original_count_h len(old_data_h) report_parts.append(f- 原始数据行数{original_count_h}) # 清洗修复行数 if headers_h_old and headers_h_new and old_data_h and new_data_h: # 找到 accept_time 字段索引 old_accept_idx -1 new_accept_idx -1 for idx, h in enumerate(headers_h_old): if accept_time in h.lower(): old_accept_idx idx break for idx, h in enumerate(headers_h_new): if accept_time in h.lower(): new_accept_idx idx break if old_accept_idx ! -1 and new_accept_idx ! -1: fixed_count count_fixed_rows( old_data_h, new_data_h, old_accept_idx, new_accept_idx ) report_parts.append(f- 清洗修复行数{fixed_count}) else: report_parts.append(- 清洗修复行数0未找到accept_time字段) # 结果预览前6行 report_parts.append(- 结果预览前6行) table_preview generate_markdown_table(headers_h_new, new_data_h, 6) report_parts.append(table_preview) else: report_parts.append(- 清洗修复行数0) report_parts.append(- 结果预览前6行无数据) # 组合报告 report_content \n\n.join(report_parts) return {baogao: report_content}
