MCP Sheet Parser

Instructions

Implementation Reference

• src/models/tools.py:102-139 (registration)

  Registration of the 'apply_changes' tool in the MCP server, including name, description, and input schema definition.
  Tool( name="apply_changes", description="将修改后的数据写回Excel/CSV文件，完成数据编辑闭环。接受parse_sheet返回的JSON格式数据（修改后）。保留原文件格式和样式，默认创建备份文件防止数据丢失。支持添加、删除、修改行和单元格数据。", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "需要写回数据的目标文件的绝对路径。" }, "table_model_json": { "type": "object", "description": "从 `parse_sheet` 工具获取并修改后的 TableModel JSON 数据。", "properties": { "sheet_name": {"type": "string"}, "headers": { "type": "array", "items": {"type": "string"} }, "rows": { "type": "array", "items": { "type": "array", "items": {} } } }, "required": ["sheet_name", "headers", "rows"] }, "create_backup": { "type": "boolean", "description": "【可选】是否在写入前创建原始文件的备份。默认为 `true`，以防意外覆盖。" } }, "required": ["file_path", "table_model_json"] } ) ]
• src/models/tools.py:356-418 (handler)

  The MCP tool handler for 'apply_changes': validates arguments, calls core_service.apply_changes, and returns structured JSON response or error.
  async def _handle_apply_changes(arguments: dict[str, Any], core_service: CoreService) -> list[TextContent]: """处理 apply_changes 工具调用。""" try: result = core_service.apply_changes( arguments["file_path"], arguments["table_model_json"], arguments.get("create_backup", True) ) response = { "success": True, "operation": "apply_changes", "result": result, "message": "数据已成功写回文件", "backup_created": arguments.get("create_backup", True) } return [TextContent( type="text", text=json.dumps(response, ensure_ascii=False, indent=2) )] except FileNotFoundError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "file_not_found", "error_message": f"文件未找到: {str(e)}", "suggestion": "请检查文件路径是否正确，确保目标文件存在" }, ensure_ascii=False, indent=2) )] except PermissionError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "permission_error", "error_message": f"权限不足: {str(e)}", "suggestion": "请检查文件权限，确保有写入文件的权限，文件可能被其他程序占用" }, ensure_ascii=False, indent=2) )] except ValueError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "invalid_data", "error_message": f"数据格式错误: {str(e)}", "suggestion": "请确保table_model_json格式正确，包含必需的字段：sheet_name, headers, rows" }, ensure_ascii=False, indent=2) )] except Exception as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "write_error", "error_message": f"写入失败: {str(e)}", "suggestion": "请检查数据格式是否与原文件兼容，或尝试关闭可能占用文件的程序" }, ensure_ascii=False, indent=2) )]
• src/core_service.py:262-323 (handler)

  Core implementation of apply_changes: validates file and JSON, creates backup, dispatches to format-specific write-back methods (_write_back_csv, _write_back_xlsx, _write_back_xls).
  def apply_changes(self, file_path: str, table_model_json: dict[str, Any], create_backup: bool = True) -> dict[str, Any]: """ 将TableModel JSON的修改应用回原始文件。 参数： file_path: 目标文件路径 table_model_json: 包含修改的JSON数据 create_backup: 是否创建备份文件 返回值： 操作结果 """ try: # 验证文件存在 path = Path(file_path) if not path.exists(): raise FileNotFoundError(f"文件不存在: {file_path}") # 验证文件格式 file_format = path.suffix.lower() supported_formats = ['.csv', '.xlsx', '.xlsm', '.xls'] if file_format not in supported_formats: if file_format == '.xlsb': raise ValueError("XLSB格式暂不支持数据写回，请转换为XLSX格式进行编辑") raise ValueError(f"Unsupported file type: {file_format}") # 创建备份文件（如果需要） backup_path = None if create_backup: backup_path = path.with_suffix(f"{path.suffix}.backup") import shutil shutil.copy2(path, backup_path) logger.info(f"已创建备份文件: {backup_path}") # 验证JSON格式 required_fields = ["sheet_name", "headers", "rows"] for field in required_fields: if field not in table_model_json: raise ValueError(f"缺少必需字段: {field}") # 实现真正的数据写回功能 changes_applied = 0 if file_format == '.csv': changes_applied = self._write_back_csv(path, table_model_json) elif file_format in ['.xlsx', '.xlsm']: changes_applied = self._write_back_xlsx(path, table_model_json) elif file_format == '.xls': changes_applied = self._write_back_xls(path, table_model_json) return { "status": "success", "message": "数据修改已成功应用", "file_path": str(path.absolute()), "backup_path": str(backup_path) if backup_path else None, "backup_created": create_backup, "changes_applied": changes_applied, "sheet_name": table_model_json.get("sheet_name"), "headers_count": len(table_model_json.get("headers", [])), "rows_count": len(table_model_json.get("rows", [])) }
• src/core_service.py:406-546 (helper)

  Helper method for writing changes back to XLSX files: clears existing data, writes headers and rows with type conversion and merged cell handling, preserves styles.
  def _write_back_xlsx(self, file_path: Path, table_model_json: dict[str, Any]) -> int: """ 将修改写回XLSX文件。 参数： file_path: XLSX文件路径 table_model_json: 包含修改的JSON数据 返回值： 应用的修改数量 """ import openpyxl headers = table_model_json["headers"] rows = table_model_json["rows"] # 打开现有的工作簿 workbook = openpyxl.load_workbook(file_path) # 根据提供的sheet_name选择正确的工作表 sheet_name_to_write = table_model_json["sheet_name"] if sheet_name_to_write in workbook.sheetnames: worksheet = workbook[sheet_name_to_write] else: raise ValueError(f"工作表 '{sheet_name_to_write}' 在文件中不存在。") # 清除现有数据（保留样式） max_row = worksheet.max_row max_col = worksheet.max_column # 清除单元格内容但保留格式 for row in range(1, max_row + 1): for col in range(1, max_col + 1): cell = worksheet.cell(row=row, column=col) # 检查是否为合并单元格，跳过MergedCell类型 try: from openpyxl.cell.cell import MergedCell if isinstance(cell, MergedCell): # 跳过被合并的单元格，因为它们的value属性是只读的 continue except ImportError: # 如果导入失败，使用字符串检查作为备选方案 if 'MergedCell' in str(type(cell)): continue cell.value = None # 写入表头 for col_idx, header in enumerate(headers, 1): cell = worksheet.cell(row=1, column=col_idx) # 检查是否为合并单元格 try: from openpyxl.cell.cell import MergedCell if isinstance(cell, MergedCell): # 跳过被合并的单元格，因为它们的value属性是只读的 # 只有合并区域的左上角单元格可以写入 logger.debug(f"跳过合并单元格 {cell.coordinate}") continue except ImportError: # 如果导入失败，使用字符串检查作为备选方案 if 'MergedCell' in str(type(cell)): logger.debug(f"跳过合并单元格 {cell.coordinate}") continue cell.value = header # 写入数据行 changes_count = 0 for row_idx, row_data in enumerate(rows, 2): # 从第2行开始（第1行是表头） for col_idx, cell_data in enumerate(row_data, 1): cell = worksheet.cell(row=row_idx, column=col_idx) # 提取单元格值 if isinstance(cell_data, dict) and 'value' in cell_data: value = cell_data['value'] else: value = cell_data # 尝试转换数值类型 # 检查是否为合并单元格（MergedCell的value属性是只读的） is_merged_cell = False try: from openpyxl.cell.cell import MergedCell if isinstance(cell, MergedCell): is_merged_cell = True except ImportError: # 如果导入失败，使用字符串检查作为备选方案 if 'MergedCell' in str(type(cell)): is_merged_cell = True if is_merged_cell: # 跳过被合并的单元格，因为它们的value属性是只读的 # 只有合并区域的左上角单元格可以写入 logger.debug(f"跳过合并单元格 {cell.coordinate}") continue # 写入值到普通单元格（此时已确认不是MergedCell） try: # 使用类型守卫确保不是 MergedCell from openpyxl.cell.cell import MergedCell if not isinstance(cell, MergedCell): if value is not None and value != "": # 改进的数值转换逻辑 if isinstance(value, str): # 尝试转换为数字，使用更安全的方法 try: # 先尝试转换为整数 if '.' not in value and 'e' not in value.lower(): cell.value = int(value) else: # 尝试转换为浮点数 cell.value = float(value) except ValueError: # 如果转换失败，保持为字符串 cell.value = str(value) elif isinstance(value, (int, float, bool)): # 保持原始数据类型 cell.value = value else: # 对于其他类型，转换为字符串 cell.value = str(value) else: cell.value = None except AttributeError as e: # 如果仍然遇到MergedCell问题，记录并跳过 if "read-only" in str(e): logger.warning(f"跳过只读单元格 {cell.coordinate}: {e}") continue else: raise changes_count += 1 # 保存工作簿 workbook.save(file_path) workbook.close() logger.info(f"XLSX文件已更新: {file_path}") return changes_count
• src/core_service.py:328-369 (helper)

  Helper method for writing changes back to CSV files: constructs rows from JSON and overwrites the file using csv.writer.
  def _write_back_csv(self, file_path: Path, table_model_json: dict[str, Any]) -> int: """ 将修改写回CSV文件。 参数： file_path: CSV文件路径 table_model_json: 包含修改的JSON数据 返回值： 应用的修改数量 """ import csv headers = table_model_json["headers"] rows = table_model_json["rows"] # 准备写入的数据 csv_rows = [] # 添加表头 csv_rows.append(headers) # 添加数据行 for row in rows: csv_row = [] for cell in row: # 提取单元格值 if isinstance(cell, dict) and 'value' in cell: value = cell['value'] else: value = str(cell) if cell is not None else "" csv_row.append(str(value) if value is not None else "") csv_rows.append(csv_row) # 写入CSV文件 with open(file_path, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.writer(csvfile) writer.writerows(csv_rows) logger.info(f"CSV文件已更新: {file_path}") return len(rows) # 返回修改的行数