mirror of
https://github.com/haris-musa/excel-mcp-server.git
synced 2025-12-26 10:07:12 +08:00
Initial commit
This commit is contained in:
21
src/excel_mcp/__main__.py
Normal file
21
src/excel_mcp/__main__.py
Normal file
@ -0,0 +1,21 @@
|
||||
import asyncio
|
||||
from .server import run_server
|
||||
|
||||
def main():
|
||||
"""Start the Excel MCP server."""
|
||||
try:
|
||||
print("Excel MCP Server")
|
||||
print("---------------")
|
||||
print("Starting server... Press Ctrl+C to exit")
|
||||
asyncio.run(run_server())
|
||||
except KeyboardInterrupt:
|
||||
print("\nShutting down server...")
|
||||
except Exception as e:
|
||||
print(f"\nError: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
print("Server stopped.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
60
src/excel_mcp/calculations.py
Normal file
60
src/excel_mcp/calculations.py
Normal file
@ -0,0 +1,60 @@
|
||||
from typing import Any
|
||||
import logging
|
||||
|
||||
from .workbook import get_or_create_workbook
|
||||
from .cell_utils import validate_cell_reference
|
||||
from .exceptions import ValidationError, CalculationError
|
||||
from .validation import validate_formula
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def apply_formula(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
cell: str,
|
||||
formula: str
|
||||
) -> dict[str, Any]:
|
||||
"""Apply any Excel formula to a cell."""
|
||||
try:
|
||||
if not validate_cell_reference(cell):
|
||||
raise ValidationError(f"Invalid cell reference: {cell}")
|
||||
|
||||
wb = get_or_create_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
sheet = wb[sheet_name]
|
||||
|
||||
# Ensure formula starts with =
|
||||
if not formula.startswith('='):
|
||||
formula = f'={formula}'
|
||||
|
||||
# Validate formula syntax
|
||||
is_valid, message = validate_formula(formula)
|
||||
if not is_valid:
|
||||
raise CalculationError(f"Invalid formula syntax: {message}")
|
||||
|
||||
try:
|
||||
# Apply formula to the cell
|
||||
cell_obj = sheet[cell]
|
||||
cell_obj.value = formula
|
||||
except Exception as e:
|
||||
raise CalculationError(f"Failed to apply formula to cell: {str(e)}")
|
||||
|
||||
try:
|
||||
wb.save(filepath)
|
||||
except Exception as e:
|
||||
raise CalculationError(f"Failed to save workbook after applying formula: {str(e)}")
|
||||
|
||||
return {
|
||||
"message": f"Applied formula '{formula}' to cell {cell}",
|
||||
"cell": cell,
|
||||
"formula": formula
|
||||
}
|
||||
|
||||
except (ValidationError, CalculationError) as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to apply formula: {e}")
|
||||
raise CalculationError(str(e))
|
||||
54
src/excel_mcp/cell_utils.py
Normal file
54
src/excel_mcp/cell_utils.py
Normal file
@ -0,0 +1,54 @@
|
||||
import re
|
||||
|
||||
from openpyxl.utils import column_index_from_string
|
||||
|
||||
def parse_cell_range(
|
||||
cell_ref: str,
|
||||
end_ref: str | None = None
|
||||
) -> tuple[int, int, int | None, int | None]:
|
||||
"""Parse Excel cell reference into row and column indices."""
|
||||
if end_ref:
|
||||
start_cell = cell_ref
|
||||
end_cell = end_ref
|
||||
else:
|
||||
start_cell = cell_ref
|
||||
end_cell = None
|
||||
|
||||
match = re.match(r"([A-Z]+)([0-9]+)", start_cell.upper())
|
||||
if not match:
|
||||
raise ValueError(f"Invalid cell reference: {start_cell}")
|
||||
col_str, row_str = match.groups()
|
||||
start_row = int(row_str)
|
||||
start_col = column_index_from_string(col_str)
|
||||
|
||||
if end_cell:
|
||||
match = re.match(r"([A-Z]+)([0-9]+)", end_cell.upper())
|
||||
if not match:
|
||||
raise ValueError(f"Invalid cell reference: {end_cell}")
|
||||
col_str, row_str = match.groups()
|
||||
end_row = int(row_str)
|
||||
end_col = column_index_from_string(col_str)
|
||||
else:
|
||||
end_row = None
|
||||
end_col = None
|
||||
|
||||
return start_row, start_col, end_row, end_col
|
||||
|
||||
def validate_cell_reference(cell_ref: str) -> bool:
|
||||
"""Validate Excel cell reference format (e.g., 'A1', 'BC123')"""
|
||||
if not cell_ref:
|
||||
return False
|
||||
|
||||
# Split into column and row parts
|
||||
col = row = ""
|
||||
for c in cell_ref:
|
||||
if c.isalpha():
|
||||
if row: # Letters after numbers not allowed
|
||||
return False
|
||||
col += c
|
||||
elif c.isdigit():
|
||||
row += c
|
||||
else:
|
||||
return False
|
||||
|
||||
return bool(col and row)
|
||||
237
src/excel_mcp/chart.py
Normal file
237
src/excel_mcp/chart.py
Normal file
@ -0,0 +1,237 @@
|
||||
from typing import Any, Optional, Dict
|
||||
import logging
|
||||
from enum import Enum
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.chart import (
|
||||
BarChart, LineChart, PieChart, ScatterChart,
|
||||
AreaChart, Reference, Series
|
||||
)
|
||||
from openpyxl.chart.label import DataLabelList
|
||||
from openpyxl.chart.legend import Legend
|
||||
from openpyxl.chart.axis import ChartLines
|
||||
from openpyxl.drawing.spreadsheet_drawing import (
|
||||
AnchorMarker, OneCellAnchor, SpreadsheetDrawing
|
||||
)
|
||||
from openpyxl.utils import column_index_from_string
|
||||
|
||||
from .cell_utils import parse_cell_range
|
||||
from .exceptions import ValidationError, ChartError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class ChartType(str, Enum):
|
||||
"""Supported chart types"""
|
||||
LINE = "line"
|
||||
BAR = "bar"
|
||||
PIE = "pie"
|
||||
SCATTER = "scatter"
|
||||
AREA = "area"
|
||||
BUBBLE = "bubble"
|
||||
STOCK = "stock"
|
||||
SURFACE = "surface"
|
||||
RADAR = "radar"
|
||||
|
||||
class ChartStyle:
|
||||
"""Chart style configuration"""
|
||||
def __init__(
|
||||
self,
|
||||
title_size: int = 14,
|
||||
title_bold: bool = True,
|
||||
axis_label_size: int = 12,
|
||||
show_legend: bool = True,
|
||||
legend_position: str = "r",
|
||||
show_data_labels: bool = True,
|
||||
grid_lines: bool = False,
|
||||
style_id: int = 2
|
||||
):
|
||||
self.title_size = title_size
|
||||
self.title_bold = title_bold
|
||||
self.axis_label_size = axis_label_size
|
||||
self.show_legend = show_legend
|
||||
self.legend_position = legend_position
|
||||
self.show_data_labels = show_data_labels
|
||||
self.grid_lines = grid_lines
|
||||
self.style_id = style_id
|
||||
|
||||
def create_chart_in_sheet(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
data_range: str,
|
||||
chart_type: str,
|
||||
target_cell: str,
|
||||
title: str = "",
|
||||
x_axis: str = "",
|
||||
y_axis: str = "",
|
||||
style: Optional[Dict] = None
|
||||
) -> dict[str, Any]:
|
||||
"""Create chart in sheet with enhanced styling options"""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
logger.error(f"Sheet '{sheet_name}' not found")
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
worksheet = wb[sheet_name]
|
||||
|
||||
# Initialize collections if they don't exist
|
||||
if not hasattr(worksheet, '_drawings'):
|
||||
worksheet._drawings = []
|
||||
if not hasattr(worksheet, '_charts'):
|
||||
worksheet._charts = []
|
||||
|
||||
# Parse the data range
|
||||
if "!" in data_range:
|
||||
range_sheet_name, cell_range = data_range.split("!")
|
||||
if range_sheet_name not in wb.sheetnames:
|
||||
logger.error(f"Sheet '{range_sheet_name}' referenced in data range not found")
|
||||
raise ValidationError(f"Sheet '{range_sheet_name}' referenced in data range not found")
|
||||
else:
|
||||
cell_range = data_range
|
||||
|
||||
try:
|
||||
start_cell, end_cell = cell_range.split(":")
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
except ValueError as e:
|
||||
logger.error(f"Invalid data range format: {e}")
|
||||
raise ValidationError(f"Invalid data range format: {str(e)}")
|
||||
|
||||
# Validate chart type
|
||||
chart_classes = {
|
||||
"line": LineChart,
|
||||
"bar": BarChart,
|
||||
"pie": PieChart,
|
||||
"scatter": ScatterChart,
|
||||
"area": AreaChart
|
||||
}
|
||||
|
||||
chart_type_lower = chart_type.lower()
|
||||
ChartClass = chart_classes.get(chart_type_lower)
|
||||
if not ChartClass:
|
||||
logger.error(f"Unsupported chart type: {chart_type}")
|
||||
raise ValidationError(
|
||||
f"Unsupported chart type: {chart_type}. "
|
||||
f"Supported types: {', '.join(chart_classes.keys())}"
|
||||
)
|
||||
|
||||
chart = ChartClass()
|
||||
|
||||
# Basic chart settings
|
||||
chart.title = title
|
||||
if hasattr(chart, "x_axis"):
|
||||
chart.x_axis.title = x_axis
|
||||
if hasattr(chart, "y_axis"):
|
||||
chart.y_axis.title = y_axis
|
||||
|
||||
try:
|
||||
# Create data references
|
||||
if chart_type_lower == "scatter":
|
||||
# For scatter charts, create series for each pair of columns
|
||||
for col in range(start_col + 1, end_col + 1):
|
||||
x_values = Reference(
|
||||
worksheet,
|
||||
min_row=start_row + 1,
|
||||
max_row=end_row,
|
||||
min_col=start_col
|
||||
)
|
||||
y_values = Reference(
|
||||
worksheet,
|
||||
min_row=start_row + 1,
|
||||
max_row=end_row,
|
||||
min_col=col
|
||||
)
|
||||
series = Series(y_values, x_values, title_from_data=True)
|
||||
chart.series.append(series)
|
||||
else:
|
||||
# For other chart types
|
||||
data = Reference(
|
||||
worksheet,
|
||||
min_row=start_row,
|
||||
max_row=end_row,
|
||||
min_col=start_col + 1,
|
||||
max_col=end_col
|
||||
)
|
||||
cats = Reference(
|
||||
worksheet,
|
||||
min_row=start_row + 1,
|
||||
max_row=end_row,
|
||||
min_col=start_col
|
||||
)
|
||||
chart.add_data(data, titles_from_data=True)
|
||||
chart.set_categories(cats)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create chart data references: {e}")
|
||||
raise ChartError(f"Failed to create chart data references: {str(e)}")
|
||||
|
||||
# Apply style if provided
|
||||
try:
|
||||
if style:
|
||||
if style.get("show_legend", True):
|
||||
chart.legend = Legend()
|
||||
chart.legend.position = style.get("legend_position", "r")
|
||||
else:
|
||||
chart.legend = None
|
||||
|
||||
if style.get("show_data_labels", False):
|
||||
chart.dataLabels = DataLabelList()
|
||||
chart.dataLabels.showVal = True
|
||||
|
||||
if style.get("grid_lines", False):
|
||||
if hasattr(chart, "x_axis"):
|
||||
chart.x_axis.majorGridlines = ChartLines()
|
||||
if hasattr(chart, "y_axis"):
|
||||
chart.y_axis.majorGridlines = ChartLines()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to apply chart style: {e}")
|
||||
raise ChartError(f"Failed to apply chart style: {str(e)}")
|
||||
|
||||
# Set chart size
|
||||
chart.width = 15
|
||||
chart.height = 7.5
|
||||
|
||||
# Create drawing and anchor
|
||||
try:
|
||||
drawing = SpreadsheetDrawing()
|
||||
drawing.chart = chart
|
||||
|
||||
# Validate target cell format
|
||||
if not target_cell or not any(c.isalpha() for c in target_cell) or not any(c.isdigit() for c in target_cell):
|
||||
raise ValidationError(f"Invalid target cell format: {target_cell}")
|
||||
|
||||
# Create anchor
|
||||
col = column_index_from_string(target_cell[0]) - 1
|
||||
row = int(target_cell[1:]) - 1
|
||||
anchor = OneCellAnchor()
|
||||
anchor._from = AnchorMarker(col=col, row=row)
|
||||
drawing.anchor = anchor
|
||||
|
||||
# Add to worksheet
|
||||
worksheet._drawings.append(drawing)
|
||||
worksheet._charts.append(chart)
|
||||
except ValueError as e:
|
||||
logger.error(f"Invalid target cell: {e}")
|
||||
raise ValidationError(f"Invalid target cell: {str(e)}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create chart drawing: {e}")
|
||||
raise ChartError(f"Failed to create chart drawing: {str(e)}")
|
||||
|
||||
try:
|
||||
wb.save(filepath)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save workbook: {e}")
|
||||
raise ChartError(f"Failed to save workbook with chart: {str(e)}")
|
||||
|
||||
return {
|
||||
"message": f"{chart_type.capitalize()} chart created successfully",
|
||||
"details": {
|
||||
"type": chart_type,
|
||||
"location": target_cell,
|
||||
"data_range": data_range
|
||||
}
|
||||
}
|
||||
|
||||
except (ValidationError, ChartError):
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error creating chart: {e}")
|
||||
raise ChartError(f"Unexpected error creating chart: {str(e)}")
|
||||
199
src/excel_mcp/data.py
Normal file
199
src/excel_mcp/data.py
Normal file
@ -0,0 +1,199 @@
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
import logging
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.styles import Font
|
||||
from openpyxl.worksheet.worksheet import Worksheet
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
from .exceptions import DataError
|
||||
from .cell_utils import parse_cell_range
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def read_excel_range(
|
||||
filepath: Path | str,
|
||||
sheet_name: str,
|
||||
start_cell: str = "A1",
|
||||
end_cell: str | None = None,
|
||||
preview_only: bool = False
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Read data from Excel range with optional preview mode"""
|
||||
try:
|
||||
wb = load_workbook(filepath, read_only=True)
|
||||
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise DataError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
ws = wb[sheet_name]
|
||||
|
||||
# Parse start cell
|
||||
if ':' in start_cell:
|
||||
start_cell, end_cell = start_cell.split(':')
|
||||
|
||||
# Get start coordinates
|
||||
try:
|
||||
start_coords = parse_cell_range(f"{start_cell}:{start_cell}")
|
||||
if not start_coords or not all(coord is not None for coord in start_coords[:2]):
|
||||
raise DataError(f"Invalid start cell reference: {start_cell}")
|
||||
start_row, start_col = start_coords[0], start_coords[1]
|
||||
except ValueError as e:
|
||||
raise DataError(f"Invalid start cell format: {str(e)}")
|
||||
|
||||
# Determine end coordinates
|
||||
if end_cell:
|
||||
try:
|
||||
end_coords = parse_cell_range(f"{end_cell}:{end_cell}")
|
||||
if not end_coords or not all(coord is not None for coord in end_coords[:2]):
|
||||
raise DataError(f"Invalid end cell reference: {end_cell}")
|
||||
end_row, end_col = end_coords[0], end_coords[1]
|
||||
except ValueError as e:
|
||||
raise DataError(f"Invalid end cell format: {str(e)}")
|
||||
else:
|
||||
# For single cell, use same coordinates
|
||||
end_row, end_col = start_row, start_col
|
||||
|
||||
# Validate range bounds
|
||||
if start_row > ws.max_row or start_col > ws.max_column:
|
||||
raise DataError(
|
||||
f"Start cell out of bounds. Sheet dimensions are "
|
||||
f"A1:{get_column_letter(ws.max_column)}{ws.max_row}"
|
||||
)
|
||||
|
||||
data = []
|
||||
# If it's a single cell or single row, just read the values directly
|
||||
if start_row == end_row:
|
||||
row_data = {}
|
||||
for col in range(start_col, end_col + 1):
|
||||
cell = ws.cell(row=start_row, column=col)
|
||||
col_name = f"Column_{col}"
|
||||
row_data[col_name] = cell.value
|
||||
if any(v is not None for v in row_data.values()):
|
||||
data.append(row_data)
|
||||
else:
|
||||
# Multiple rows - use header row
|
||||
headers = []
|
||||
for col in range(start_col, end_col + 1):
|
||||
cell_value = ws.cell(row=start_row, column=col).value
|
||||
headers.append(str(cell_value) if cell_value is not None else f"Column_{col}")
|
||||
|
||||
# Get data rows
|
||||
max_rows = min(start_row + 5, end_row) if preview_only else end_row
|
||||
for row in range(start_row + 1, max_rows + 1):
|
||||
row_data = {}
|
||||
for col, header in enumerate(headers, start=start_col):
|
||||
cell = ws.cell(row=row, column=col)
|
||||
row_data[header] = cell.value
|
||||
if any(v is not None for v in row_data.values()):
|
||||
data.append(row_data)
|
||||
|
||||
wb.close()
|
||||
return data
|
||||
except DataError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to read Excel range: {e}")
|
||||
raise DataError(str(e))
|
||||
|
||||
def write_data(
|
||||
filepath: str,
|
||||
sheet_name: str | None,
|
||||
data: list[dict[str, Any]] | None,
|
||||
start_cell: str = "A1",
|
||||
write_headers: bool = True,
|
||||
) -> dict[str, str]:
|
||||
"""Write data to Excel sheet with workbook handling"""
|
||||
try:
|
||||
if not data:
|
||||
raise DataError("No data provided to write")
|
||||
|
||||
wb = load_workbook(filepath)
|
||||
|
||||
# If no sheet specified, use active sheet
|
||||
if not sheet_name:
|
||||
sheet_name = wb.active.title
|
||||
elif sheet_name not in wb.sheetnames:
|
||||
wb.create_sheet(sheet_name)
|
||||
|
||||
ws = wb[sheet_name]
|
||||
|
||||
# Validate start cell
|
||||
try:
|
||||
start_coords = parse_cell_range(start_cell)
|
||||
if not start_coords or not all(coord is not None for coord in start_coords[:2]):
|
||||
raise DataError(f"Invalid start cell reference: {start_cell}")
|
||||
except ValueError as e:
|
||||
raise DataError(f"Invalid start cell format: {str(e)}")
|
||||
|
||||
if len(data) > 0:
|
||||
# Check if first row of data contains headers
|
||||
first_row = data[0]
|
||||
has_headers = all(
|
||||
isinstance(value, str) and value.strip() == key.strip()
|
||||
for key, value in first_row.items()
|
||||
)
|
||||
|
||||
# If first row contains headers, skip it when write_headers is True
|
||||
if has_headers and write_headers:
|
||||
data = data[1:]
|
||||
|
||||
_write_data_to_worksheet(ws, data, start_cell, write_headers)
|
||||
|
||||
wb.save(filepath)
|
||||
wb.close()
|
||||
|
||||
return {"message": f"Data written to {sheet_name}", "active_sheet": sheet_name}
|
||||
except DataError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to write data: {e}")
|
||||
raise DataError(str(e))
|
||||
|
||||
def _write_data_to_worksheet(
|
||||
worksheet: Worksheet,
|
||||
data: list[dict[str, Any]],
|
||||
start_cell: str = "A1",
|
||||
write_headers: bool = True,
|
||||
) -> None:
|
||||
"""Write data to worksheet - internal helper function"""
|
||||
try:
|
||||
if not data:
|
||||
raise DataError("No data provided to write")
|
||||
|
||||
try:
|
||||
start_coords = parse_cell_range(start_cell)
|
||||
if not start_coords or not all(x is not None for x in start_coords[:2]):
|
||||
raise DataError(f"Invalid start cell reference: {start_cell}")
|
||||
start_row, start_col = start_coords[0], start_coords[1]
|
||||
except ValueError as e:
|
||||
raise DataError(f"Invalid start cell format: {str(e)}")
|
||||
|
||||
# Validate data structure
|
||||
if not all(isinstance(row, dict) for row in data):
|
||||
raise DataError("All data rows must be dictionaries")
|
||||
|
||||
# Write headers if requested
|
||||
headers = list(data[0].keys())
|
||||
if write_headers:
|
||||
for i, header in enumerate(headers):
|
||||
cell = worksheet.cell(row=start_row, column=start_col + i)
|
||||
cell.value = header
|
||||
cell.font = Font(bold=True)
|
||||
start_row += 1 # Move start row down if headers were written
|
||||
|
||||
# Write data
|
||||
for i, row_dict in enumerate(data):
|
||||
if not all(h in row_dict for h in headers):
|
||||
raise DataError(f"Row {i+1} is missing required headers")
|
||||
for j, header in enumerate(headers):
|
||||
cell = worksheet.cell(row=start_row + i, column=start_col + j)
|
||||
cell.value = row_dict.get(header, "")
|
||||
except DataError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to write worksheet data: {e}")
|
||||
raise DataError(str(e))
|
||||
35
src/excel_mcp/exceptions.py
Normal file
35
src/excel_mcp/exceptions.py
Normal file
@ -0,0 +1,35 @@
|
||||
class ExcelMCPError(Exception):
|
||||
"""Base exception for Excel MCP errors."""
|
||||
pass
|
||||
|
||||
class WorkbookError(ExcelMCPError):
|
||||
"""Raised when workbook operations fail."""
|
||||
pass
|
||||
|
||||
class SheetError(ExcelMCPError):
|
||||
"""Raised when sheet operations fail."""
|
||||
pass
|
||||
|
||||
class DataError(ExcelMCPError):
|
||||
"""Raised when data operations fail."""
|
||||
pass
|
||||
|
||||
class ValidationError(ExcelMCPError):
|
||||
"""Raised when validation fails."""
|
||||
pass
|
||||
|
||||
class FormattingError(ExcelMCPError):
|
||||
"""Raised when formatting operations fail."""
|
||||
pass
|
||||
|
||||
class CalculationError(ExcelMCPError):
|
||||
"""Raised when formula calculations fail."""
|
||||
pass
|
||||
|
||||
class PivotError(ExcelMCPError):
|
||||
"""Raised when pivot table operations fail."""
|
||||
pass
|
||||
|
||||
class ChartError(ExcelMCPError):
|
||||
"""Raised when chart operations fail."""
|
||||
pass
|
||||
249
src/excel_mcp/formatting.py
Normal file
249
src/excel_mcp/formatting.py
Normal file
@ -0,0 +1,249 @@
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
|
||||
from openpyxl.styles import (
|
||||
PatternFill, Border, Side, Alignment, Protection, Font,
|
||||
Color
|
||||
)
|
||||
from openpyxl.formatting.rule import (
|
||||
ColorScaleRule, DataBarRule, IconSetRule,
|
||||
FormulaRule, CellIsRule
|
||||
)
|
||||
|
||||
from .workbook import get_or_create_workbook
|
||||
from .cell_utils import parse_cell_range, validate_cell_reference
|
||||
from .exceptions import ValidationError, FormattingError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def format_range(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str = None,
|
||||
bold: bool = False,
|
||||
italic: bool = False,
|
||||
underline: bool = False,
|
||||
font_size: int = None,
|
||||
font_color: str = None,
|
||||
bg_color: str = None,
|
||||
border_style: str = None,
|
||||
border_color: str = None,
|
||||
number_format: str = None,
|
||||
alignment: str = None,
|
||||
wrap_text: bool = False,
|
||||
merge_cells: bool = False,
|
||||
protection: Dict[str, Any] = None,
|
||||
conditional_format: Dict[str, Any] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Apply formatting to a range of cells.
|
||||
|
||||
This function handles all Excel formatting operations including:
|
||||
- Font properties (bold, italic, size, color, etc.)
|
||||
- Cell fill/background color
|
||||
- Borders (style and color)
|
||||
- Number formatting
|
||||
- Alignment and text wrapping
|
||||
- Cell merging
|
||||
- Protection
|
||||
- Conditional formatting
|
||||
|
||||
Args:
|
||||
filepath: Path to Excel file
|
||||
sheet_name: Name of worksheet
|
||||
start_cell: Starting cell reference
|
||||
end_cell: Optional ending cell reference
|
||||
bold: Whether to make text bold
|
||||
italic: Whether to make text italic
|
||||
underline: Whether to underline text
|
||||
font_size: Font size in points
|
||||
font_color: Font color (hex code)
|
||||
bg_color: Background color (hex code)
|
||||
border_style: Border style (thin, medium, thick, double)
|
||||
border_color: Border color (hex code)
|
||||
number_format: Excel number format string
|
||||
alignment: Text alignment (left, center, right, justify)
|
||||
wrap_text: Whether to wrap text
|
||||
merge_cells: Whether to merge the range
|
||||
protection: Cell protection settings
|
||||
conditional_format: Conditional formatting rules
|
||||
|
||||
Returns:
|
||||
Dictionary with operation status
|
||||
"""
|
||||
try:
|
||||
# Validate cell references
|
||||
if not validate_cell_reference(start_cell):
|
||||
raise ValidationError(f"Invalid start cell reference: {start_cell}")
|
||||
|
||||
if end_cell and not validate_cell_reference(end_cell):
|
||||
raise ValidationError(f"Invalid end cell reference: {end_cell}")
|
||||
|
||||
wb = get_or_create_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
sheet = wb[sheet_name]
|
||||
|
||||
# Get cell range coordinates
|
||||
try:
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
except ValueError as e:
|
||||
raise ValidationError(f"Invalid cell range: {str(e)}")
|
||||
|
||||
# If no end cell specified, use start cell coordinates
|
||||
if end_row is None:
|
||||
end_row = start_row
|
||||
if end_col is None:
|
||||
end_col = start_col
|
||||
|
||||
# Apply font formatting
|
||||
font_args = {
|
||||
"bold": bold,
|
||||
"italic": italic,
|
||||
"underline": 'single' if underline else None,
|
||||
}
|
||||
if font_size is not None:
|
||||
font_args["size"] = font_size
|
||||
if font_color is not None:
|
||||
try:
|
||||
# Ensure color has FF prefix for full opacity
|
||||
font_color = font_color if font_color.startswith('FF') else f'FF{font_color}'
|
||||
font_args["color"] = Color(rgb=font_color)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid font color: {str(e)}")
|
||||
font = Font(**font_args)
|
||||
|
||||
# Apply fill
|
||||
fill = None
|
||||
if bg_color is not None:
|
||||
try:
|
||||
# Ensure color has FF prefix for full opacity
|
||||
bg_color = bg_color if bg_color.startswith('FF') else f'FF{bg_color}'
|
||||
fill = PatternFill(
|
||||
start_color=Color(rgb=bg_color),
|
||||
end_color=Color(rgb=bg_color),
|
||||
fill_type='solid'
|
||||
)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid background color: {str(e)}")
|
||||
|
||||
# Apply borders
|
||||
border = None
|
||||
if border_style is not None:
|
||||
try:
|
||||
border_color = border_color if border_color else "000000"
|
||||
border_color = border_color if border_color.startswith('FF') else f'FF{border_color}'
|
||||
side = Side(
|
||||
style=border_style,
|
||||
color=Color(rgb=border_color)
|
||||
)
|
||||
border = Border(
|
||||
left=side,
|
||||
right=side,
|
||||
top=side,
|
||||
bottom=side
|
||||
)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid border settings: {str(e)}")
|
||||
|
||||
# Apply alignment
|
||||
align = None
|
||||
if alignment is not None or wrap_text:
|
||||
try:
|
||||
align = Alignment(
|
||||
horizontal=alignment,
|
||||
vertical='center',
|
||||
wrap_text=wrap_text
|
||||
)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid alignment settings: {str(e)}")
|
||||
|
||||
# Apply protection
|
||||
protect = None
|
||||
if protection is not None:
|
||||
try:
|
||||
protect = Protection(**protection)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid protection settings: {str(e)}")
|
||||
|
||||
# Apply formatting to range
|
||||
for row in range(start_row, end_row + 1):
|
||||
for col in range(start_col, end_col + 1):
|
||||
cell = sheet.cell(row=row, column=col)
|
||||
cell.font = font
|
||||
if fill is not None:
|
||||
cell.fill = fill
|
||||
if border is not None:
|
||||
cell.border = border
|
||||
if align is not None:
|
||||
cell.alignment = align
|
||||
if protect is not None:
|
||||
cell.protection = protect
|
||||
if number_format is not None:
|
||||
cell.number_format = number_format
|
||||
|
||||
# Merge cells if requested
|
||||
if merge_cells and end_cell:
|
||||
try:
|
||||
range_str = f"{start_cell}:{end_cell}"
|
||||
sheet.merge_cells(range_str)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Failed to merge cells: {str(e)}")
|
||||
|
||||
# Apply conditional formatting
|
||||
if conditional_format is not None:
|
||||
range_str = f"{start_cell}:{end_cell}" if end_cell else start_cell
|
||||
rule_type = conditional_format.get('type')
|
||||
if not rule_type:
|
||||
raise FormattingError("Conditional format type not specified")
|
||||
|
||||
params = conditional_format.get('params', {})
|
||||
|
||||
# Handle fill parameter for cell_is rule
|
||||
if rule_type == 'cell_is' and 'fill' in params:
|
||||
fill_params = params['fill']
|
||||
if isinstance(fill_params, dict):
|
||||
try:
|
||||
fill_color = fill_params.get('fgColor', 'FFC7CE') # Default to light red
|
||||
fill_color = fill_color if fill_color.startswith('FF') else f'FF{fill_color}'
|
||||
params['fill'] = PatternFill(
|
||||
start_color=fill_color,
|
||||
end_color=fill_color,
|
||||
fill_type='solid'
|
||||
)
|
||||
except ValueError as e:
|
||||
raise FormattingError(f"Invalid conditional format fill color: {str(e)}")
|
||||
|
||||
try:
|
||||
if rule_type == 'color_scale':
|
||||
rule = ColorScaleRule(**params)
|
||||
elif rule_type == 'data_bar':
|
||||
rule = DataBarRule(**params)
|
||||
elif rule_type == 'icon_set':
|
||||
rule = IconSetRule(**params)
|
||||
elif rule_type == 'formula':
|
||||
rule = FormulaRule(**params)
|
||||
elif rule_type == 'cell_is':
|
||||
rule = CellIsRule(**params)
|
||||
else:
|
||||
raise FormattingError(f"Invalid conditional format type: {rule_type}")
|
||||
|
||||
sheet.conditional_formatting.add(range_str, rule)
|
||||
except Exception as e:
|
||||
raise FormattingError(f"Failed to apply conditional formatting: {str(e)}")
|
||||
|
||||
wb.save(filepath)
|
||||
|
||||
range_str = f"{start_cell}:{end_cell}" if end_cell else start_cell
|
||||
return {
|
||||
"message": f"Applied formatting to range {range_str}",
|
||||
"range": range_str
|
||||
}
|
||||
|
||||
except (ValidationError, FormattingError) as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to apply formatting: {e}")
|
||||
raise FormattingError(str(e))
|
||||
271
src/excel_mcp/pivot.py
Normal file
271
src/excel_mcp/pivot.py
Normal file
@ -0,0 +1,271 @@
|
||||
from typing import Any
|
||||
import uuid
|
||||
import logging
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.utils import get_column_letter
|
||||
from openpyxl.worksheet.table import Table, TableStyleInfo
|
||||
from openpyxl.styles import Font
|
||||
|
||||
from .data import read_excel_range
|
||||
from .cell_utils import parse_cell_range
|
||||
from .exceptions import ValidationError, PivotError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def create_pivot_table(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
data_range: str,
|
||||
rows: list[str],
|
||||
values: list[str],
|
||||
columns: list[str] | None = None,
|
||||
agg_func: str = "sum"
|
||||
) -> dict[str, Any]:
|
||||
"""Create pivot table in sheet using Excel table functionality
|
||||
|
||||
Args:
|
||||
filepath: Path to Excel file
|
||||
sheet_name: Name of worksheet containing source data
|
||||
data_range: Source data range reference
|
||||
target_cell: Cell reference for pivot table position
|
||||
rows: Fields for row labels
|
||||
values: Fields for values
|
||||
columns: Optional fields for column labels
|
||||
agg_func: Aggregation function (sum, count, average, max, min)
|
||||
|
||||
Returns:
|
||||
Dictionary with status message and pivot table dimensions
|
||||
"""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
# Parse ranges
|
||||
if ':' not in data_range:
|
||||
raise ValidationError("Data range must be in format 'A1:B2'")
|
||||
|
||||
try:
|
||||
start_cell, end_cell = data_range.split(':')
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
except ValueError as e:
|
||||
raise ValidationError(f"Invalid data range format: {str(e)}")
|
||||
|
||||
if end_row is None or end_col is None:
|
||||
raise ValidationError("Invalid data range format: missing end coordinates")
|
||||
|
||||
# Create range string
|
||||
data_range_str = f"{get_column_letter(start_col)}{start_row}:{get_column_letter(end_col)}{end_row}"
|
||||
|
||||
# Read source data
|
||||
try:
|
||||
data = read_excel_range(filepath, sheet_name, start_cell, end_cell)
|
||||
if not data:
|
||||
raise PivotError("No data found in range")
|
||||
except Exception as e:
|
||||
raise PivotError(f"Failed to read source data: {str(e)}")
|
||||
|
||||
# Validate aggregation function
|
||||
valid_agg_funcs = ["sum", "average", "count", "min", "max"]
|
||||
if agg_func.lower() not in valid_agg_funcs:
|
||||
raise ValidationError(
|
||||
f"Invalid aggregation function. Must be one of: {', '.join(valid_agg_funcs)}"
|
||||
)
|
||||
|
||||
# Clean up field names by removing aggregation suffixes
|
||||
def clean_field_name(field: str) -> str:
|
||||
field = str(field).strip()
|
||||
for suffix in [" (sum)", " (average)", " (count)", " (min)", " (max)"]:
|
||||
if field.lower().endswith(suffix):
|
||||
return field[:-len(suffix)]
|
||||
return field
|
||||
|
||||
# Validate field names exist in data
|
||||
if data:
|
||||
first_row = data[0]
|
||||
available_fields = {clean_field_name(str(header)).lower() for header in first_row.keys()}
|
||||
|
||||
for field_list, field_type in [(rows, "row"), (values, "value")]:
|
||||
for field in field_list:
|
||||
if clean_field_name(str(field)).lower() not in available_fields:
|
||||
raise ValidationError(
|
||||
f"Invalid {field_type} field '{field}'. "
|
||||
f"Available fields: {', '.join(sorted(available_fields))}"
|
||||
)
|
||||
|
||||
if columns:
|
||||
for field in columns:
|
||||
if clean_field_name(str(field)).lower() not in available_fields:
|
||||
raise ValidationError(
|
||||
f"Invalid column field '{field}'. "
|
||||
f"Available fields: {', '.join(sorted(available_fields))}"
|
||||
)
|
||||
|
||||
# Skip header row if it matches our fields
|
||||
if all(
|
||||
any(clean_field_name(str(header)).lower() == clean_field_name(str(field)).lower()
|
||||
for field in rows + values)
|
||||
for header in first_row.keys()
|
||||
):
|
||||
data = data[1:]
|
||||
|
||||
# Clean up row and value field names
|
||||
cleaned_rows = [clean_field_name(field) for field in rows]
|
||||
cleaned_values = [clean_field_name(field) for field in values]
|
||||
|
||||
# Create pivot sheet
|
||||
pivot_sheet_name = f"{sheet_name}_pivot"
|
||||
if pivot_sheet_name in wb.sheetnames:
|
||||
wb.remove(wb[pivot_sheet_name])
|
||||
pivot_ws = wb.create_sheet(pivot_sheet_name)
|
||||
|
||||
# Write headers
|
||||
current_row = 1
|
||||
current_col = 1
|
||||
|
||||
# Write row field headers
|
||||
for field in cleaned_rows:
|
||||
cell = pivot_ws.cell(row=current_row, column=current_col, value=field)
|
||||
cell.font = Font(bold=True)
|
||||
current_col += 1
|
||||
|
||||
# Write value field headers
|
||||
for field in cleaned_values:
|
||||
cell = pivot_ws.cell(row=current_row, column=current_col, value=f"{field} ({agg_func})")
|
||||
cell.font = Font(bold=True)
|
||||
current_col += 1
|
||||
|
||||
# Get unique values for each row field
|
||||
field_values = {}
|
||||
for field in cleaned_rows:
|
||||
all_values = []
|
||||
for record in data:
|
||||
value = str(record.get(field, ''))
|
||||
all_values.append(value)
|
||||
field_values[field] = sorted(set(all_values))
|
||||
|
||||
# Generate all combinations of row field values
|
||||
row_combinations = _get_combinations(field_values)
|
||||
|
||||
# Calculate table dimensions for formatting
|
||||
total_rows = len(row_combinations) + 1 # +1 for header
|
||||
total_cols = len(cleaned_rows) + len(cleaned_values)
|
||||
|
||||
# Write data rows
|
||||
current_row = 2
|
||||
for combo in row_combinations:
|
||||
# Write row field values
|
||||
col = 1
|
||||
for field in cleaned_rows:
|
||||
pivot_ws.cell(row=current_row, column=col, value=combo[field])
|
||||
col += 1
|
||||
|
||||
# Filter data for current combination
|
||||
filtered_data = _filter_data(data, combo, {})
|
||||
|
||||
# Calculate and write aggregated values
|
||||
for value_field in cleaned_values:
|
||||
try:
|
||||
value = _aggregate_values(filtered_data, value_field, agg_func)
|
||||
pivot_ws.cell(row=current_row, column=col, value=value)
|
||||
except Exception as e:
|
||||
raise PivotError(f"Failed to aggregate values for field '{value_field}': {str(e)}")
|
||||
col += 1
|
||||
|
||||
current_row += 1
|
||||
|
||||
# Create a table for the pivot data
|
||||
try:
|
||||
pivot_range = f"A1:{get_column_letter(total_cols)}{total_rows}"
|
||||
pivot_table = Table(
|
||||
displayName=f"PivotTable_{uuid.uuid4().hex[:8]}",
|
||||
ref=pivot_range
|
||||
)
|
||||
style = TableStyleInfo(
|
||||
name="TableStyleMedium9",
|
||||
showFirstColumn=False,
|
||||
showLastColumn=False,
|
||||
showRowStripes=True,
|
||||
showColumnStripes=True
|
||||
)
|
||||
pivot_table.tableStyleInfo = style
|
||||
pivot_ws.add_table(pivot_table)
|
||||
except Exception as e:
|
||||
raise PivotError(f"Failed to create pivot table formatting: {str(e)}")
|
||||
|
||||
try:
|
||||
wb.save(filepath)
|
||||
except Exception as e:
|
||||
raise PivotError(f"Failed to save workbook: {str(e)}")
|
||||
|
||||
return {
|
||||
"message": "Summary table created successfully",
|
||||
"details": {
|
||||
"source_range": data_range_str,
|
||||
"pivot_sheet": pivot_sheet_name,
|
||||
"rows": cleaned_rows,
|
||||
"columns": columns or [],
|
||||
"values": cleaned_values,
|
||||
"aggregation": agg_func
|
||||
}
|
||||
}
|
||||
|
||||
except (ValidationError, PivotError) as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create pivot table: {e}")
|
||||
raise PivotError(str(e))
|
||||
|
||||
|
||||
def _get_combinations(field_values: dict[str, set]) -> list[dict]:
|
||||
"""Get all combinations of field values."""
|
||||
result = [{}]
|
||||
for field, values in list(field_values.items()): # Convert to list to avoid runtime changes
|
||||
new_result = []
|
||||
for combo in result:
|
||||
for value in sorted(values): # Sort for consistent ordering
|
||||
new_combo = combo.copy()
|
||||
new_combo[field] = value
|
||||
new_result.append(new_combo)
|
||||
result = new_result
|
||||
return result
|
||||
|
||||
|
||||
def _filter_data(data: list[dict], row_filters: dict, col_filters: dict) -> list[dict]:
|
||||
"""Filter data based on row and column filters."""
|
||||
result = []
|
||||
for record in data:
|
||||
matches = True
|
||||
for field, value in row_filters.items():
|
||||
if record.get(field) != value:
|
||||
matches = False
|
||||
break
|
||||
for field, value in col_filters.items():
|
||||
if record.get(field) != value:
|
||||
matches = False
|
||||
break
|
||||
if matches:
|
||||
result.append(record)
|
||||
return result
|
||||
|
||||
|
||||
def _aggregate_values(data: list[dict], field: str, agg_func: str) -> float:
|
||||
"""Aggregate values using the specified function."""
|
||||
values = [record[field] for record in data if field in record and isinstance(record[field], (int, float))]
|
||||
if not values:
|
||||
return 0
|
||||
|
||||
if agg_func == "sum":
|
||||
return sum(values)
|
||||
elif agg_func == "average":
|
||||
return sum(values) / len(values)
|
||||
elif agg_func == "count":
|
||||
return len(values)
|
||||
elif agg_func == "min":
|
||||
return min(values)
|
||||
elif agg_func == "max":
|
||||
return max(values)
|
||||
else:
|
||||
return sum(values) # Default to sum
|
||||
491
src/excel_mcp/server.py
Normal file
491
src/excel_mcp/server.py
Normal file
@ -0,0 +1,491 @@
|
||||
import logging
|
||||
import sys
|
||||
import os
|
||||
from typing import Any, List, Dict
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
# Import exceptions
|
||||
from excel_mcp.exceptions import (
|
||||
ValidationError,
|
||||
WorkbookError,
|
||||
SheetError,
|
||||
DataError,
|
||||
FormattingError,
|
||||
CalculationError,
|
||||
PivotError,
|
||||
ChartError
|
||||
)
|
||||
|
||||
# Import from excel_mcp package with consistent _impl suffixes
|
||||
from excel_mcp.validation import (
|
||||
validate_formula_in_cell_operation as validate_formula_impl,
|
||||
validate_range_in_sheet_operation as validate_range_impl
|
||||
)
|
||||
from excel_mcp.chart import create_chart_in_sheet as create_chart_impl
|
||||
from excel_mcp.workbook import get_workbook_info
|
||||
from excel_mcp.data import write_data
|
||||
from excel_mcp.pivot import create_pivot_table as create_pivot_table_impl
|
||||
from excel_mcp.sheet import (
|
||||
copy_sheet,
|
||||
delete_sheet,
|
||||
rename_sheet,
|
||||
merge_range,
|
||||
unmerge_range,
|
||||
)
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.StreamHandler(sys.stdout),
|
||||
logging.FileHandler("excel-mcp.log")
|
||||
],
|
||||
force=True
|
||||
)
|
||||
|
||||
logger = logging.getLogger("excel-mcp")
|
||||
|
||||
# Get Excel files path from environment or use default
|
||||
EXCEL_FILES_PATH = os.environ.get("EXCEL_FILES_PATH", "./excel_files")
|
||||
|
||||
# Create the directory if it doesn't exist
|
||||
os.makedirs(EXCEL_FILES_PATH, exist_ok=True)
|
||||
|
||||
# Initialize FastMCP server
|
||||
mcp = FastMCP(
|
||||
"excel-mcp",
|
||||
version="0.1.0",
|
||||
description="Excel MCP Server for manipulating Excel files",
|
||||
dependencies=["openpyxl>=3.1.2"],
|
||||
env_vars={
|
||||
"EXCEL_FILES_PATH": {
|
||||
"description": "Path to Excel files directory",
|
||||
"required": False,
|
||||
"default": EXCEL_FILES_PATH
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
def get_excel_path(filename: str) -> str:
|
||||
"""Get full path to Excel file.
|
||||
|
||||
Args:
|
||||
filename: Name of Excel file
|
||||
|
||||
Returns:
|
||||
Full path to Excel file
|
||||
"""
|
||||
# If filename is already an absolute path, return it
|
||||
if os.path.isabs(filename):
|
||||
return filename
|
||||
|
||||
# Use the configured Excel files path
|
||||
return os.path.join(EXCEL_FILES_PATH, filename)
|
||||
|
||||
@mcp.tool()
|
||||
def apply_formula(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
cell: str,
|
||||
formula: str,
|
||||
) -> str:
|
||||
"""Apply Excel formula to cell."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
# First validate the formula
|
||||
validation = validate_formula_impl(full_path, sheet_name, cell, formula)
|
||||
if isinstance(validation, dict) and "error" in validation:
|
||||
return f"Error: {validation['error']}"
|
||||
|
||||
# If valid, apply the formula
|
||||
from excel_mcp.calculations import apply_formula as apply_formula_impl
|
||||
result = apply_formula_impl(full_path, sheet_name, cell, formula)
|
||||
return result["message"]
|
||||
except (ValidationError, CalculationError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error applying formula: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def validate_formula_syntax(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
cell: str,
|
||||
formula: str,
|
||||
) -> str:
|
||||
"""Validate Excel formula syntax without applying it."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = validate_formula_impl(full_path, sheet_name, cell, formula)
|
||||
return result["message"]
|
||||
except (ValidationError, CalculationError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error validating formula: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def format_range(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str = None,
|
||||
bold: bool = False,
|
||||
italic: bool = False,
|
||||
underline: bool = False,
|
||||
font_size: int = None,
|
||||
font_color: str = None,
|
||||
bg_color: str = None,
|
||||
border_style: str = None,
|
||||
border_color: str = None,
|
||||
number_format: str = None,
|
||||
alignment: str = None,
|
||||
wrap_text: bool = False,
|
||||
merge_cells: bool = False,
|
||||
protection: Dict[str, Any] = None,
|
||||
conditional_format: Dict[str, Any] = None
|
||||
) -> str:
|
||||
"""Apply formatting to a range of cells."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.formatting import format_range as format_range_func
|
||||
|
||||
result = format_range_func(
|
||||
filepath=full_path,
|
||||
sheet_name=sheet_name,
|
||||
start_cell=start_cell,
|
||||
end_cell=end_cell,
|
||||
bold=bold,
|
||||
italic=italic,
|
||||
underline=underline,
|
||||
font_size=font_size,
|
||||
font_color=font_color,
|
||||
bg_color=bg_color,
|
||||
border_style=border_style,
|
||||
border_color=border_color,
|
||||
number_format=number_format,
|
||||
alignment=alignment,
|
||||
wrap_text=wrap_text,
|
||||
merge_cells=merge_cells,
|
||||
protection=protection,
|
||||
conditional_format=conditional_format
|
||||
)
|
||||
return "Range formatted successfully"
|
||||
except (ValidationError, FormattingError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error formatting range: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def read_data_from_excel(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str = "A1",
|
||||
end_cell: str = None,
|
||||
preview_only: bool = False
|
||||
) -> str:
|
||||
"""Read data from Excel worksheet."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.data import read_excel_range
|
||||
result = read_excel_range(full_path, sheet_name, start_cell, end_cell, preview_only)
|
||||
if not result:
|
||||
return "No data found in specified range"
|
||||
# Convert the list of dicts to a formatted string
|
||||
data_str = "\n".join([str(row) for row in result])
|
||||
return data_str
|
||||
except Exception as e:
|
||||
logger.error(f"Error reading data: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def write_data_to_excel(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
data: List[Dict],
|
||||
start_cell: str = "A1",
|
||||
write_headers: bool = True,
|
||||
) -> str:
|
||||
"""Write data to Excel worksheet."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = write_data(full_path, sheet_name, data, start_cell, write_headers)
|
||||
return result["message"]
|
||||
except (ValidationError, DataError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error writing data: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def create_workbook(filepath: str) -> str:
|
||||
"""Create new Excel workbook."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.workbook import create_workbook as create_workbook_impl
|
||||
result = create_workbook_impl(full_path)
|
||||
return f"Created workbook at {full_path}"
|
||||
except WorkbookError as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating workbook: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def create_worksheet(filepath: str, sheet_name: str) -> str:
|
||||
"""Create new worksheet in workbook."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.workbook import create_sheet as create_worksheet_impl
|
||||
result = create_worksheet_impl(full_path, sheet_name)
|
||||
return result["message"]
|
||||
except (ValidationError, WorkbookError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating worksheet: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def create_chart(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
data_range: str,
|
||||
chart_type: str,
|
||||
target_cell: str,
|
||||
title: str = "",
|
||||
x_axis: str = "",
|
||||
y_axis: str = ""
|
||||
) -> str:
|
||||
"""Create chart in worksheet."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = create_chart_impl(
|
||||
filepath=full_path,
|
||||
sheet_name=sheet_name,
|
||||
data_range=data_range,
|
||||
chart_type=chart_type,
|
||||
target_cell=target_cell,
|
||||
title=title,
|
||||
x_axis=x_axis,
|
||||
y_axis=y_axis
|
||||
)
|
||||
return result["message"]
|
||||
except (ValidationError, ChartError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating chart: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def create_pivot_table(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
data_range: str,
|
||||
target_cell: str,
|
||||
rows: List[str],
|
||||
values: List[str],
|
||||
columns: List[str] = None,
|
||||
agg_func: str = "mean"
|
||||
) -> str:
|
||||
"""Create pivot table in worksheet."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = create_pivot_table_impl(
|
||||
filepath=full_path,
|
||||
sheet_name=sheet_name,
|
||||
data_range=data_range,
|
||||
target_cell=target_cell,
|
||||
rows=rows,
|
||||
values=values,
|
||||
columns=columns or [],
|
||||
agg_func=agg_func
|
||||
)
|
||||
return result["message"]
|
||||
except (ValidationError, PivotError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating pivot table: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def copy_worksheet(
|
||||
filepath: str,
|
||||
source_sheet: str,
|
||||
target_sheet: str
|
||||
) -> str:
|
||||
"""Copy worksheet within workbook."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = copy_sheet(full_path, source_sheet, target_sheet)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error copying worksheet: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def delete_worksheet(
|
||||
filepath: str,
|
||||
sheet_name: str
|
||||
) -> str:
|
||||
"""Delete worksheet from workbook."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = delete_sheet(full_path, sheet_name)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting worksheet: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def rename_worksheet(
|
||||
filepath: str,
|
||||
old_name: str,
|
||||
new_name: str
|
||||
) -> str:
|
||||
"""Rename worksheet in workbook."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = rename_sheet(full_path, old_name, new_name)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error renaming worksheet: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def get_workbook_metadata(
|
||||
filepath: str,
|
||||
include_ranges: bool = False
|
||||
) -> str:
|
||||
"""Get metadata about workbook including sheets, ranges, etc."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = get_workbook_info(full_path, include_ranges=include_ranges)
|
||||
return str(result)
|
||||
except WorkbookError as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting workbook metadata: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def merge_cells(filepath: str, sheet_name: str, start_cell: str, end_cell: str) -> str:
|
||||
"""Merge a range of cells."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = merge_range(full_path, sheet_name, start_cell, end_cell)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error merging cells: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def unmerge_cells(filepath: str, sheet_name: str, start_cell: str, end_cell: str) -> str:
|
||||
"""Unmerge a range of cells."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
result = unmerge_range(full_path, sheet_name, start_cell, end_cell)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error unmerging cells: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def copy_range(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
source_start: str,
|
||||
source_end: str,
|
||||
target_start: str,
|
||||
target_sheet: str = None
|
||||
) -> str:
|
||||
"""Copy a range of cells to another location."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.sheet import copy_range_operation
|
||||
result = copy_range_operation(
|
||||
full_path,
|
||||
sheet_name,
|
||||
source_start,
|
||||
source_end,
|
||||
target_start,
|
||||
target_sheet
|
||||
)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error copying range: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def delete_range(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str,
|
||||
shift_direction: str = "up"
|
||||
) -> str:
|
||||
"""Delete a range of cells and shift remaining cells."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
from excel_mcp.sheet import delete_range_operation
|
||||
result = delete_range_operation(
|
||||
full_path,
|
||||
sheet_name,
|
||||
start_cell,
|
||||
end_cell,
|
||||
shift_direction
|
||||
)
|
||||
return result["message"]
|
||||
except (ValidationError, SheetError) as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting range: {e}")
|
||||
raise
|
||||
|
||||
@mcp.tool()
|
||||
def validate_excel_range(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str = None
|
||||
) -> str:
|
||||
"""Validate if a range exists and is properly formatted."""
|
||||
try:
|
||||
full_path = get_excel_path(filepath)
|
||||
range_str = start_cell if not end_cell else f"{start_cell}:{end_cell}"
|
||||
result = validate_range_impl(full_path, sheet_name, range_str)
|
||||
return result["message"]
|
||||
except ValidationError as e:
|
||||
return f"Error: {str(e)}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error validating range: {e}")
|
||||
raise
|
||||
|
||||
async def run_server():
|
||||
"""Run the Excel MCP server."""
|
||||
try:
|
||||
logger.info(f"Starting Excel MCP server (files directory: {EXCEL_FILES_PATH})")
|
||||
await mcp.run_sse_async()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Server stopped by user")
|
||||
await mcp.shutdown()
|
||||
except Exception as e:
|
||||
logger.error(f"Server failed: {e}")
|
||||
raise
|
||||
finally:
|
||||
logger.info("Server shutdown complete")
|
||||
327
src/excel_mcp/sheet.py
Normal file
327
src/excel_mcp/sheet.py
Normal file
@ -0,0 +1,327 @@
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.worksheet.worksheet import Worksheet
|
||||
from openpyxl.utils import get_column_letter
|
||||
from openpyxl.styles import Font, Border, PatternFill, Side
|
||||
|
||||
from .cell_utils import parse_cell_range, validate_cell_reference
|
||||
from .exceptions import SheetError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def copy_sheet(filepath: str, source_sheet: str, target_sheet: str) -> dict[str, Any]:
|
||||
"""Copy a worksheet within the same workbook."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if source_sheet not in wb.sheetnames:
|
||||
raise SheetError(f"Source sheet '{source_sheet}' not found")
|
||||
|
||||
if target_sheet in wb.sheetnames:
|
||||
raise SheetError(f"Target sheet '{target_sheet}' already exists")
|
||||
|
||||
source = wb[source_sheet]
|
||||
target = wb.copy_worksheet(source)
|
||||
target.title = target_sheet
|
||||
|
||||
wb.save(filepath)
|
||||
return {"message": f"Sheet '{source_sheet}' copied to '{target_sheet}'"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to copy sheet: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def delete_sheet(filepath: str, sheet_name: str) -> dict[str, Any]:
|
||||
"""Delete a worksheet from the workbook."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
if len(wb.sheetnames) == 1:
|
||||
raise SheetError("Cannot delete the only sheet in workbook")
|
||||
|
||||
del wb[sheet_name]
|
||||
wb.save(filepath)
|
||||
return {"message": f"Sheet '{sheet_name}' deleted"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete sheet: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def rename_sheet(filepath: str, old_name: str, new_name: str) -> dict[str, Any]:
|
||||
"""Rename a worksheet."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if old_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{old_name}' not found")
|
||||
|
||||
if new_name in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{new_name}' already exists")
|
||||
|
||||
sheet = wb[old_name]
|
||||
sheet.title = new_name
|
||||
wb.save(filepath)
|
||||
return {"message": f"Sheet renamed from '{old_name}' to '{new_name}'"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to rename sheet: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def format_range_string(start_row: int, start_col: int, end_row: int, end_col: int) -> str:
|
||||
"""Format range string from row and column indices."""
|
||||
return f"{get_column_letter(start_col)}{start_row}:{get_column_letter(end_col)}{end_row}"
|
||||
|
||||
def copy_range(
|
||||
source_ws: Worksheet,
|
||||
target_ws: Worksheet,
|
||||
source_range: str,
|
||||
target_start: str | None = None,
|
||||
) -> None:
|
||||
"""Copy range from source worksheet to target worksheet."""
|
||||
# Parse source range
|
||||
if ':' in source_range:
|
||||
source_start, source_end = source_range.split(':')
|
||||
else:
|
||||
source_start = source_range
|
||||
source_end = None
|
||||
|
||||
src_start_row, src_start_col, src_end_row, src_end_col = parse_cell_range(
|
||||
source_start, source_end
|
||||
)
|
||||
|
||||
if src_end_row is None:
|
||||
src_end_row = src_start_row
|
||||
src_end_col = src_start_col
|
||||
|
||||
if target_start is None:
|
||||
target_start = source_start
|
||||
|
||||
tgt_start_row, tgt_start_col, _, _ = parse_cell_range(target_start)
|
||||
|
||||
for i, row in enumerate(range(src_start_row, src_end_row + 1)):
|
||||
for j, col in enumerate(range(src_start_col, src_end_col + 1)):
|
||||
source_cell = source_ws.cell(row=row, column=col)
|
||||
target_cell = target_ws.cell(row=tgt_start_row + i, column=tgt_start_col + j)
|
||||
|
||||
target_cell.value = source_cell.value
|
||||
|
||||
try:
|
||||
# Copy font
|
||||
font_kwargs = {}
|
||||
if hasattr(source_cell.font, 'name'):
|
||||
font_kwargs['name'] = source_cell.font.name
|
||||
if hasattr(source_cell.font, 'size'):
|
||||
font_kwargs['size'] = source_cell.font.size
|
||||
if hasattr(source_cell.font, 'bold'):
|
||||
font_kwargs['bold'] = source_cell.font.bold
|
||||
if hasattr(source_cell.font, 'italic'):
|
||||
font_kwargs['italic'] = source_cell.font.italic
|
||||
if hasattr(source_cell.font, 'color'):
|
||||
font_color = None
|
||||
if source_cell.font.color:
|
||||
font_color = source_cell.font.color.rgb
|
||||
font_kwargs['color'] = font_color
|
||||
target_cell.font = Font(**font_kwargs)
|
||||
|
||||
# Copy border
|
||||
new_border = Border()
|
||||
for side in ['left', 'right', 'top', 'bottom']:
|
||||
source_side = getattr(source_cell.border, side)
|
||||
if source_side and source_side.style:
|
||||
side_color = source_side.color.rgb if source_side.color else None
|
||||
setattr(new_border, side, Side(
|
||||
style=source_side.style,
|
||||
color=side_color
|
||||
))
|
||||
target_cell.border = new_border
|
||||
|
||||
# Copy fill
|
||||
if hasattr(source_cell, 'fill'):
|
||||
fill_kwargs = {'patternType': source_cell.fill.patternType}
|
||||
if hasattr(source_cell.fill, 'fgColor') and source_cell.fill.fgColor:
|
||||
fg_color = None
|
||||
if hasattr(source_cell.fill.fgColor, 'rgb'):
|
||||
fg_color = source_cell.fill.fgColor.rgb
|
||||
fill_kwargs['fgColor'] = fg_color
|
||||
if hasattr(source_cell.fill, 'bgColor') and source_cell.fill.bgColor:
|
||||
bg_color = None
|
||||
if hasattr(source_cell.fill.bgColor, 'rgb'):
|
||||
bg_color = source_cell.fill.bgColor.rgb
|
||||
fill_kwargs['bgColor'] = bg_color
|
||||
target_cell.fill = PatternFill(**fill_kwargs)
|
||||
|
||||
# Copy number format and alignment
|
||||
if source_cell.number_format:
|
||||
target_cell.number_format = source_cell.number_format
|
||||
if source_cell.alignment:
|
||||
target_cell.alignment = source_cell.alignment
|
||||
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
def delete_range(worksheet: Worksheet, start_cell: str, end_cell: str | None = None) -> None:
|
||||
"""Delete contents and formatting of a range."""
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
|
||||
if end_row is None:
|
||||
end_row = start_row
|
||||
end_col = start_col
|
||||
|
||||
for row in range(start_row, end_row + 1):
|
||||
for col in range(start_col, end_col + 1):
|
||||
cell = worksheet.cell(row=row, column=col)
|
||||
cell.value = None
|
||||
cell.font = Font()
|
||||
cell.border = Border()
|
||||
cell.fill = PatternFill()
|
||||
cell.number_format = "General"
|
||||
cell.alignment = None
|
||||
|
||||
def merge_range(filepath: str, sheet_name: str, start_cell: str, end_cell: str) -> dict[str, Any]:
|
||||
"""Merge a range of cells."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
|
||||
if end_row is None or end_col is None:
|
||||
raise SheetError("Both start and end cells must be specified for merging")
|
||||
|
||||
range_string = format_range_string(start_row, start_col, end_row, end_col)
|
||||
worksheet = wb[sheet_name]
|
||||
worksheet.merge_cells(range_string)
|
||||
wb.save(filepath)
|
||||
return {"message": f"Range '{range_string}' merged in sheet '{sheet_name}'"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to merge range: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def unmerge_range(filepath: str, sheet_name: str, start_cell: str, end_cell: str) -> dict[str, Any]:
|
||||
"""Unmerge a range of cells."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
worksheet = wb[sheet_name]
|
||||
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
|
||||
if end_row is None or end_col is None:
|
||||
raise SheetError("Both start and end cells must be specified for unmerging")
|
||||
|
||||
range_string = format_range_string(start_row, start_col, end_row, end_col)
|
||||
|
||||
# Check if range is actually merged
|
||||
merged_ranges = worksheet.merged_cells.ranges
|
||||
target_range = range_string.upper()
|
||||
|
||||
if not any(str(merged_range).upper() == target_range for merged_range in merged_ranges):
|
||||
raise SheetError(f"Range '{range_string}' is not merged")
|
||||
|
||||
worksheet.unmerge_cells(range_string)
|
||||
wb.save(filepath)
|
||||
return {"message": f"Range '{range_string}' unmerged successfully"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to unmerge range: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def copy_range_operation(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
source_start: str,
|
||||
source_end: str | None,
|
||||
target_start: str,
|
||||
target_sheet: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Copy a range of cells to another location."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
source_ws = wb[sheet_name]
|
||||
target_ws = wb[target_sheet] if target_sheet else source_ws
|
||||
|
||||
if target_sheet and target_sheet not in wb.sheetnames:
|
||||
raise SheetError(f"Target sheet '{target_sheet}' not found")
|
||||
|
||||
source_range = f"{source_start}:{source_end}" if source_end else source_start
|
||||
|
||||
# Validate source range
|
||||
try:
|
||||
end_row, end_col = parse_cell_range(source_start, source_end)
|
||||
if end_row and end_row > source_ws.max_row:
|
||||
raise SheetError(f"End row {end_row} out of bounds (1-{source_ws.max_row})")
|
||||
if end_col and end_col > source_ws.max_column:
|
||||
raise SheetError(f"End column {end_col} out of bounds (1-{source_ws.max_column})")
|
||||
except ValueError as e:
|
||||
raise SheetError(f"Invalid range: {str(e)}")
|
||||
|
||||
# Validate target cell
|
||||
try:
|
||||
validate_cell_reference(target_start)
|
||||
except ValueError as e:
|
||||
raise SheetError(f"Invalid target cell: {str(e)}")
|
||||
|
||||
copy_range(source_ws, target_ws, source_range, target_start)
|
||||
wb.save(filepath)
|
||||
|
||||
return {"message": f"Range copied successfully"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to copy range: {e}")
|
||||
raise SheetError(str(e))
|
||||
|
||||
def delete_range_operation(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Delete a range of cells and shift remaining cells."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise SheetError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
worksheet = wb[sheet_name]
|
||||
|
||||
# Validate range
|
||||
try:
|
||||
end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
if end_row and end_row > worksheet.max_row:
|
||||
raise SheetError(f"End row {end_row} out of bounds (1-{worksheet.max_row})")
|
||||
if end_col and end_col > worksheet.max_column:
|
||||
raise SheetError(f"End column {end_col} out of bounds (1-{worksheet.max_column})")
|
||||
except ValueError as e:
|
||||
raise SheetError(f"Invalid range: {str(e)}")
|
||||
|
||||
delete_range(worksheet, start_cell, end_cell)
|
||||
wb.save(filepath)
|
||||
|
||||
return {"message": f"Range deleted successfully"}
|
||||
except SheetError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete range: {e}")
|
||||
raise SheetError(str(e))
|
||||
235
src/excel_mcp/validation.py
Normal file
235
src/excel_mcp/validation.py
Normal file
@ -0,0 +1,235 @@
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.utils import get_column_letter
|
||||
from openpyxl.worksheet.worksheet import Worksheet
|
||||
|
||||
from .cell_utils import parse_cell_range, validate_cell_reference
|
||||
from .exceptions import ValidationError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def validate_formula_in_cell_operation(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
cell: str,
|
||||
formula: str
|
||||
) -> dict[str, Any]:
|
||||
"""Validate Excel formula before writing"""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
if not validate_cell_reference(cell):
|
||||
raise ValidationError(f"Invalid cell reference: {cell}")
|
||||
|
||||
# First validate the provided formula's syntax
|
||||
is_valid, message = validate_formula(formula)
|
||||
if not is_valid:
|
||||
raise ValidationError(f"Invalid formula syntax: {message}")
|
||||
|
||||
# Additional validation for cell references in formula
|
||||
cell_refs = re.findall(r'[A-Z]+[0-9]+(?::[A-Z]+[0-9]+)?', formula)
|
||||
for ref in cell_refs:
|
||||
if ':' in ref: # Range reference
|
||||
start, end = ref.split(':')
|
||||
if not (validate_cell_reference(start) and validate_cell_reference(end)):
|
||||
raise ValidationError(f"Invalid cell range reference in formula: {ref}")
|
||||
else: # Single cell reference
|
||||
if not validate_cell_reference(ref):
|
||||
raise ValidationError(f"Invalid cell reference in formula: {ref}")
|
||||
|
||||
# Now check if there's a formula in the cell and compare
|
||||
sheet = wb[sheet_name]
|
||||
cell_obj = sheet[cell]
|
||||
current_formula = cell_obj.value
|
||||
|
||||
# If cell has a formula (starts with =)
|
||||
if isinstance(current_formula, str) and current_formula.startswith('='):
|
||||
if formula.startswith('='):
|
||||
if current_formula != formula:
|
||||
return {
|
||||
"message": "Formula is valid but doesn't match cell content",
|
||||
"valid": True,
|
||||
"matches": False,
|
||||
"cell": cell,
|
||||
"provided_formula": formula,
|
||||
"current_formula": current_formula
|
||||
}
|
||||
else:
|
||||
if current_formula != f"={formula}":
|
||||
return {
|
||||
"message": "Formula is valid but doesn't match cell content",
|
||||
"valid": True,
|
||||
"matches": False,
|
||||
"cell": cell,
|
||||
"provided_formula": formula,
|
||||
"current_formula": current_formula
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"message": "Formula is valid and matches cell content",
|
||||
"valid": True,
|
||||
"matches": True,
|
||||
"cell": cell,
|
||||
"formula": formula
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"message": "Formula is valid but cell contains no formula",
|
||||
"valid": True,
|
||||
"matches": False,
|
||||
"cell": cell,
|
||||
"provided_formula": formula,
|
||||
"current_content": str(current_formula) if current_formula else ""
|
||||
}
|
||||
|
||||
except ValidationError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to validate formula: {e}")
|
||||
raise ValidationError(str(e))
|
||||
|
||||
def validate_range_in_sheet_operation(
|
||||
filepath: str,
|
||||
sheet_name: str,
|
||||
start_cell: str,
|
||||
end_cell: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate if a range exists in a worksheet and return data range info."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
if sheet_name not in wb.sheetnames:
|
||||
raise ValidationError(f"Sheet '{sheet_name}' not found")
|
||||
|
||||
worksheet = wb[sheet_name]
|
||||
|
||||
# Get actual data dimensions
|
||||
data_max_row = worksheet.max_row
|
||||
data_max_col = worksheet.max_column
|
||||
|
||||
# Validate range
|
||||
try:
|
||||
start_row, start_col, end_row, end_col = parse_cell_range(start_cell, end_cell)
|
||||
except ValueError as e:
|
||||
raise ValidationError(f"Invalid range: {str(e)}")
|
||||
|
||||
# If end not specified, use start
|
||||
if end_row is None:
|
||||
end_row = start_row
|
||||
if end_col is None:
|
||||
end_col = start_col
|
||||
|
||||
# Validate bounds against maximum possible Excel limits
|
||||
is_valid, message = validate_range_bounds(
|
||||
worksheet, start_row, start_col, end_row, end_col
|
||||
)
|
||||
if not is_valid:
|
||||
raise ValidationError(message)
|
||||
|
||||
range_str = f"{start_cell}" if end_cell is None else f"{start_cell}:{end_cell}"
|
||||
data_range_str = f"A1:{get_column_letter(data_max_col)}{data_max_row}"
|
||||
|
||||
# Check if range is within data or extends beyond
|
||||
extends_beyond_data = (
|
||||
end_row > data_max_row or
|
||||
end_col > data_max_col
|
||||
)
|
||||
|
||||
return {
|
||||
"message": (
|
||||
f"Range '{range_str}' is valid. "
|
||||
f"Sheet contains data in range '{data_range_str}'"
|
||||
),
|
||||
"valid": True,
|
||||
"range": range_str,
|
||||
"data_range": data_range_str,
|
||||
"extends_beyond_data": extends_beyond_data,
|
||||
"data_dimensions": {
|
||||
"max_row": data_max_row,
|
||||
"max_col": data_max_col,
|
||||
"max_col_letter": get_column_letter(data_max_col)
|
||||
}
|
||||
}
|
||||
except ValidationError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to validate range: {e}")
|
||||
raise ValidationError(str(e))
|
||||
|
||||
def validate_formula(formula: str) -> tuple[bool, str]:
|
||||
"""Validate Excel formula syntax and safety"""
|
||||
if not formula.startswith("="):
|
||||
return False, "Formula must start with '='"
|
||||
|
||||
# Remove the '=' prefix for validation
|
||||
formula = formula[1:]
|
||||
|
||||
# Check for balanced parentheses
|
||||
parens = 0
|
||||
for c in formula:
|
||||
if c == "(":
|
||||
parens += 1
|
||||
elif c == ")":
|
||||
parens -= 1
|
||||
if parens < 0:
|
||||
return False, "Unmatched closing parenthesis"
|
||||
|
||||
if parens > 0:
|
||||
return False, "Unclosed parenthesis"
|
||||
|
||||
# Basic function name validation
|
||||
func_pattern = r"([A-Z]+)\("
|
||||
funcs = re.findall(func_pattern, formula)
|
||||
unsafe_funcs = {"INDIRECT", "HYPERLINK", "WEBSERVICE", "DGET", "RTD"}
|
||||
|
||||
for func in funcs:
|
||||
if func in unsafe_funcs:
|
||||
return False, f"Unsafe function: {func}"
|
||||
|
||||
return True, "Formula is valid"
|
||||
|
||||
|
||||
def validate_range_bounds(
|
||||
worksheet: Worksheet,
|
||||
start_row: int,
|
||||
start_col: int,
|
||||
end_row: int | None = None,
|
||||
end_col: int | None = None,
|
||||
) -> tuple[bool, str]:
|
||||
"""Validate that cell range is within worksheet bounds"""
|
||||
max_row = worksheet.max_row
|
||||
max_col = worksheet.max_column
|
||||
|
||||
try:
|
||||
# Check start cell bounds
|
||||
if start_row < 1 or start_row > max_row:
|
||||
return False, f"Start row {start_row} out of bounds (1-{max_row})"
|
||||
if start_col < 1 or start_col > max_col:
|
||||
return False, (
|
||||
f"Start column {get_column_letter(start_col)} "
|
||||
f"out of bounds (A-{get_column_letter(max_col)})"
|
||||
)
|
||||
|
||||
# If end cell specified, check its bounds
|
||||
if end_row is not None and end_col is not None:
|
||||
if end_row < start_row:
|
||||
return False, "End row cannot be before start row"
|
||||
if end_col < start_col:
|
||||
return False, "End column cannot be before start column"
|
||||
if end_row > max_row:
|
||||
return False, f"End row {end_row} out of bounds (1-{max_row})"
|
||||
if end_col > max_col:
|
||||
return False, (
|
||||
f"End column {get_column_letter(end_col)} "
|
||||
f"out of bounds (A-{get_column_letter(max_col)})"
|
||||
)
|
||||
|
||||
return True, "Range is valid"
|
||||
except Exception as e:
|
||||
return False, f"Invalid range: {e!s}"
|
||||
96
src/excel_mcp/workbook.py
Normal file
96
src/excel_mcp/workbook.py
Normal file
@ -0,0 +1,96 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from openpyxl import Workbook, load_workbook
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
from .exceptions import WorkbookError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def create_workbook(filepath: str, sheet_name: str = "Sheet1") -> dict[str, Any]:
|
||||
"""Create a new Excel workbook with optional custom sheet name"""
|
||||
try:
|
||||
wb = Workbook()
|
||||
# Rename default sheet
|
||||
if "Sheet" in wb.sheetnames:
|
||||
sheet = wb["Sheet"]
|
||||
sheet.title = sheet_name
|
||||
else:
|
||||
wb.create_sheet(sheet_name)
|
||||
|
||||
path = Path(filepath)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
wb.save(str(path))
|
||||
return {
|
||||
"message": f"Created workbook: {filepath}",
|
||||
"active_sheet": sheet_name,
|
||||
"workbook": wb
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create workbook: {e}")
|
||||
raise WorkbookError(f"Failed to create workbook: {e!s}")
|
||||
|
||||
def get_or_create_workbook(filepath: str) -> Workbook:
|
||||
"""Get existing workbook or create new one if it doesn't exist"""
|
||||
try:
|
||||
return load_workbook(filepath)
|
||||
except FileNotFoundError:
|
||||
return create_workbook(filepath)["workbook"]
|
||||
|
||||
def create_sheet(filepath: str, sheet_name: str) -> dict:
|
||||
"""Create a new worksheet in the workbook if it doesn't exist."""
|
||||
try:
|
||||
wb = load_workbook(filepath)
|
||||
|
||||
# Check if sheet already exists
|
||||
if sheet_name in wb.sheetnames:
|
||||
raise WorkbookError(f"Sheet {sheet_name} already exists")
|
||||
|
||||
# Create new sheet
|
||||
wb.create_sheet(sheet_name)
|
||||
wb.save(filepath)
|
||||
wb.close()
|
||||
return {"message": f"Sheet {sheet_name} created successfully"}
|
||||
except WorkbookError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create sheet: {e}")
|
||||
raise WorkbookError(str(e))
|
||||
|
||||
def get_workbook_info(filepath: str, include_ranges: bool = False) -> dict[str, Any]:
|
||||
"""Get metadata about workbook including sheets, ranges, etc."""
|
||||
try:
|
||||
path = Path(filepath)
|
||||
if not path.exists():
|
||||
raise WorkbookError(f"File not found: {filepath}")
|
||||
|
||||
wb = load_workbook(filepath, read_only=True)
|
||||
|
||||
info = {
|
||||
"filename": path.name,
|
||||
"sheets": wb.sheetnames,
|
||||
"size": path.stat().st_size,
|
||||
"modified": path.stat().st_mtime
|
||||
}
|
||||
|
||||
if include_ranges:
|
||||
# Add used ranges for each sheet
|
||||
ranges = {}
|
||||
for sheet_name in wb.sheetnames:
|
||||
ws = wb[sheet_name]
|
||||
if ws.max_row > 0 and ws.max_column > 0:
|
||||
ranges[sheet_name] = f"A1:{get_column_letter(ws.max_column)}{ws.max_row}"
|
||||
info["used_ranges"] = ranges
|
||||
|
||||
wb.close()
|
||||
return info
|
||||
|
||||
except WorkbookError as e:
|
||||
logger.error(str(e))
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get workbook info: {e}")
|
||||
raise WorkbookError(str(e))
|
||||
Reference in New Issue
Block a user