MCP Sheet Parser

Instructions

Implementation Reference

• src/core_service.py:262-327 (handler)

  Core handler implementation of the 'apply_changes' tool. Validates input, creates backup, dispatches to format-specific write-back methods (_write_back_csv, _write_back_xlsx, _write_back_xls), and returns structured result.
  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", [])) } except Exception as e: logger.error(f"应用修改失败: {e}") raise
• src/models/tools.py:102-138 (schema)

  Input schema definition for the 'apply_changes' tool, including parameters file_path, table_model_json (with sheet_name, headers, rows), and optional create_backup.
  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:27-163 (registration)

  Tool registration via @server.list_tools() including apply_changes Tool object, and dispatch in @server.call_tool() handle_call_tool with elif for 'apply_changes' calling _handle_apply_changes.
  @server.list_tools() async def handle_list_tools() -> list[Tool]: return [ Tool( name="convert_to_html", description="将Excel/CSV文件转换为可在浏览器中查看的HTML文件。保留原始样式、颜色、字体等格式。支持多工作表文件，可选择特定工作表或转换全部。大文件可使用分页功能。返回结构化JSON，包含成功状态、生成的文件信息和转换摘要。", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "源表格文件的绝对路径，支持 .csv, .xlsx, .xls, .xlsb, .xlsm 格式。" }, "output_path": { "type": "string", "description": "输出HTML文件的路径。如果留空，将在源文件目录中生成一个同名的 .html 文件。" }, "sheet_name": { "type": "string", "description": "【可选】要转换的单个工作表的名称。如果留空，将转换文件中的所有工作表。" }, "page_size": { "type": "integer", "description": "【可选】分页时每页显示的行数。默认为100行。用于控制大型文件转换后HTML的单页大小。" }, "page_number": { "type": "integer", "description": "【可选】要查看的页码，从1开始。默认为1。用于浏览大型文件的特定页面。" }, "header_rows": { "type": "integer", "description": "【可选】将文件顶部的指定行数视为表头。默认为 1。" } }, "required": ["file_path"] } ), Tool( name="parse_sheet", description="解析Excel/CSV文件为结构化JSON数据。默认返回文件概览信息（行数、列数、数据类型、前几行预览），避免上下文过载。LLM可通过参数控制是否获取完整数据、样式信息等。适合数据分析和处理，修改后可用apply_changes写回。", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "目标表格文件的绝对路径，支持 .csv, .xlsx, .xls, .xlsb, .xlsm 格式。" }, "sheet_name": { "type": "string", "description": "【可选】要解析的工作表名称。如果留空，使用第一个工作表。" }, "range_string": { "type": "string", "description": "【可选】单元格范围，如'A1:D10'。指定范围时会返回该范围的完整数据。" }, "include_full_data": { "type": "boolean", "description": "【可选，默认false】是否返回完整数据。false时只返回概览和预览，true时返回所有行数据。大文件建议先查看概览。" }, "include_styles": { "type": "boolean", "description": "【可选，默认false】是否包含样式信息（字体、颜色、边框等）。样式信息会显著增加数据量。" }, "preview_rows": { "type": "integer", "description": "【可选，默认5】预览行数。当include_full_data为false时，返回前N行作为数据预览。" }, "max_rows": { "type": "integer", "description": "【可选】最大返回行数。用于限制大文件的数据量，超出部分会被截断并提示。" } }, "required": ["file_path"] } ), 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"] } ) ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: """处理工具调用请求。""" try: if name == "convert_to_html": return await _handle_convert_to_html(arguments, core_service) elif name == "parse_sheet": return await _handle_parse_sheet(arguments, core_service) elif name == "apply_changes": return await _handle_apply_changes(arguments, core_service) else: return [TextContent( type="text", text=f"未知工具: {name}" )] except Exception as e: logger.error(f"工具调用失败 {name}: {e}") return [TextContent( type="text", text=f"错误: {str(e)}" )]
• src/models/tools.py:356-419 (helper)

  Wrapper handler for apply_changes tool call: extracts arguments, calls core_service.apply_changes, formats JSON response with success/error handling.
  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:406-547 (helper)

  Key helper method for writing changes back to XLSX files: loads workbook with openpyxl, clears data (preserves styles), writes headers and rows handling merged cells and type conversion, saves file.
  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