import { execFile } from 'child_process'; import { promisify } from 'util'; import path from 'path'; import fs from 'fs'; import { ToolResult } from '../types.js'; import { getWorkspacePath } from '../config/paths.js'; const execFileAsync = promisify(execFile); // ─── Python scripts ──────────────────────────────────────────────────────────── const READ_SCRIPT = ` import sys, json try: import openpyxl except ImportError: print(json.dumps({"error": "openpyxl_missing"})) sys.exit(1) xlsx_path = sys.argv[1] sheet_name = sys.argv[2] if len(sys.argv) > 2 and sys.argv[2] != '__ALL__' else None cell_range = sys.argv[3] if len(sys.argv) > 3 and sys.argv[3] != '__NONE__' else None max_rows = int(sys.argv[4]) if len(sys.argv) > 4 else 500 wb = openpyxl.load_workbook(xlsx_path, read_only=True, data_only=True) sheet_names = wb.sheetnames if sheet_name is None: # Return sheet list + preview of first sheet ws = wb.active target_sheets = [ws.title] else: if sheet_name not in sheet_names: print(json.dumps({"error": f"Sheet '{sheet_name}' not found. Available: {sheet_names}"})) sys.exit(1) ws = wb[sheet_name] target_sheets = [sheet_name] results = [] for sname in target_sheets: ws = wb[sname] if cell_range: cells = ws[cell_range] if not hasattr(cells, '__iter__') or isinstance(cells, openpyxl.cell.cell.Cell): cells = [[cells]] elif not hasattr(cells[0], '__iter__'): cells = [cells] rows = [[str(c.value) if c.value is not None else '' for c in row] for row in cells] else: rows = [] for i, row in enumerate(ws.iter_rows(values_only=True)): if i >= max_rows: rows.append([f'... ({ws.max_row - max_rows} more rows truncated)']) break rows.append([str(v) if v is not None else '' for v in row]) # Strip trailing empty rows while rows and all(c == '' for c in rows[-1]): rows.pop() results.append({"sheet": sname, "rows": rows, "total_rows": ws.max_row, "total_cols": ws.max_column}) print(json.dumps({"sheets": sheet_names, "data": results})) `; const WRITE_SCRIPT = ` import sys, json, os try: import openpyxl from openpyxl.styles import Font, PatternFill, Alignment except ImportError: print(json.dumps({"error": "openpyxl_missing"})) sys.exit(1) spec_path = sys.argv[1] with open(spec_path) as f: spec = json.load(f) xlsx_path = spec['path'] sheet_name = spec.get('sheet', 'Sheet1') data = spec.get('data', []) headers = spec.get('headers', []) start_cell = spec.get('start_cell', 'A1') mode = spec.get('mode', 'create') # create | append | overwrite_sheet # Load or create workbook if mode != 'create' and os.path.exists(xlsx_path): wb = openpyxl.load_workbook(xlsx_path) if sheet_name in wb.sheetnames: if mode == 'overwrite_sheet': del wb[sheet_name] ws = wb.create_sheet(sheet_name) else: ws = wb[sheet_name] # append: find last row start_cell = f'A{ws.max_row + 1}' else: ws = wb.create_sheet(sheet_name) else: wb = openpyxl.Workbook() ws = wb.active ws.title = sheet_name # Parse start cell import re m = re.match(r'([A-Za-z]+)([0-9]+)', start_cell) col_letter = m.group(1).upper() if m else 'A' start_row = int(m.group(2)) if m else 1 start_col = openpyxl.utils.column_index_from_string(col_letter) # Write headers row_offset = 0 if headers: for ci, h in enumerate(headers): cell = ws.cell(row=start_row, column=start_col + ci, value=h) cell.font = Font(bold=True) cell.fill = PatternFill('solid', fgColor='D9E1F2') row_offset = 1 # Write data for ri, row in enumerate(data): for ci, val in enumerate(row): ws.cell(row=start_row + row_offset + ri, column=start_col + ci, value=val) # Auto-fit column widths (approximate) for col in ws.columns: max_len = 0 col_letter2 = openpyxl.utils.get_column_letter(col[0].column) for cell in col: try: if cell.value: max_len = max(max_len, len(str(cell.value))) except: pass ws.column_dimensions[col_letter2].width = min(max(max_len + 2, 8), 50) os.makedirs(os.path.dirname(os.path.abspath(xlsx_path)), exist_ok=True) wb.save(xlsx_path) rel = os.path.relpath(xlsx_path) print(json.dumps({"success": True, "path": xlsx_path, "sheet": sheet_name, "rows_written": len(data)})) `; // ─── Helpers ────────────────────────────────────────────────────────────────── function formatTable(rows: string[][]): string { if (rows.length === 0) return '(empty)'; const widths = rows[0].map((_, ci) => Math.min(40, Math.max(...rows.map(r => (r[ci] ?? '').length))) ); return rows .map(row => row.map((cell, ci) => (cell ?? '').padEnd(widths[ci])).join(' ')) .join('\n'); } // ─── Tools ──────────────────────────────────────────────────────────────────── export const excelReadTool = { name: 'excel_read', description: 'Read data from an Excel (.xlsx) file. Returns sheet names, row data as a formatted table. Supports reading a specific sheet, cell range, and row limit.', schema: { path: 'Path to the .xlsx file (absolute or relative to workspace)', sheet: 'Sheet name to read (default: active/first sheet)', range: 'Cell range to read, e.g. "A1:F20" (default: entire sheet)', max_rows: 'Maximum rows to return (default: 500)', }, jsonSchema: { type: 'object', properties: { path: { type: 'string', description: 'Path to .xlsx file' }, sheet: { type: 'string', description: 'Sheet name (default: first sheet)' }, range: { type: 'string', description: 'Cell range e.g. "A1:F20"' }, max_rows: { type: 'number', description: 'Max rows to return (default: 500)' }, }, required: ['path'], additionalProperties: false, }, execute: async (args: any): Promise => { const filePath = String(args?.path || '').trim(); if (!filePath) return { success: false, error: 'path is required' }; const workspacePath = getWorkspacePath(args); const resolved = path.isAbsolute(filePath) ? filePath : path.resolve(workspacePath, filePath); if (!fs.existsSync(resolved)) { return { success: false, error: `File not found: ${resolved}` }; } const ext = path.extname(resolved).toLowerCase(); if (ext !== '.xlsx') { return { success: false, error: 'Only .xlsx files are supported (not .xls or .csv)' }; } const sheetArg = args?.sheet ? String(args.sheet) : '__ALL__'; const rangeArg = args?.range ? String(args.range) : '__NONE__'; const maxRows = args?.max_rows ? Math.min(5000, Math.max(1, Number(args.max_rows))) : 500; let result: any; try { const { stdout } = await execFileAsync( 'python3', ['-c', READ_SCRIPT, resolved, sheetArg, rangeArg, String(maxRows)], { maxBuffer: 20 * 1024 * 1024, timeout: 30_000 } ); result = JSON.parse(stdout.trim()); } catch (err: any) { return { success: false, error: `excel_read failed: ${err.message}` }; } if (result.error) { if (result.error === 'openpyxl_missing') { return { success: false, error: 'openpyxl not installed. Run: pip3 install openpyxl' }; } return { success: false, error: result.error }; } const lines: string[] = [`Sheets: ${result.sheets.join(', ')}\n`]; for (const block of result.data) { lines.push(`=== Sheet: ${block.sheet} (${block.total_rows} rows × ${block.total_cols} cols) ===`); lines.push(formatTable(block.rows)); } return { success: true, stdout: lines.join('\n'), data: { sheets: result.sheets, sheetCount: result.sheets.length }, }; }, }; export const excelWriteTool = { name: 'excel_write', description: 'Create or update an Excel (.xlsx) file. Writes data rows with optional bold headers. Supports creating new files, appending rows, or overwriting a specific sheet.', schema: { path: 'Output path for .xlsx file (absolute or relative to workspace)', sheet: 'Sheet name (default: "Sheet1")', headers: 'Array of column header strings (written in bold)', data: 'Array of rows, each row is an array of values', start_cell: 'Top-left cell to start writing, e.g. "A1" (default: "A1")', mode: 'create | append | overwrite_sheet (default: create)', }, jsonSchema: { type: 'object', properties: { path: { type: 'string', description: 'Output .xlsx path' }, sheet: { type: 'string', description: 'Sheet name (default: Sheet1)' }, headers: { type: 'array', items: { type: 'string' }, description: 'Column headers (bold)' }, data: { type: 'array', items: { type: 'array' }, description: 'Rows of data (array of arrays)' }, start_cell: { type: 'string', description: 'Start cell e.g. "A1"' }, mode: { type: 'string', enum: ['create', 'append', 'overwrite_sheet'], description: 'Write mode' }, }, required: ['path', 'data'], additionalProperties: false, }, execute: async (args: any): Promise => { const filePath = String(args?.path || '').trim(); if (!filePath) return { success: false, error: 'path is required' }; if (!Array.isArray(args?.data)) return { success: false, error: 'data must be an array of rows' }; const workspacePath = getWorkspacePath(args); const resolved = path.isAbsolute(filePath) ? filePath : path.resolve(workspacePath, filePath); if (path.extname(resolved).toLowerCase() !== '.xlsx') { return { success: false, error: 'Output file must have .xlsx extension' }; } const spec = { path: resolved, sheet: args?.sheet || 'Sheet1', headers: args?.headers || [], data: args?.data || [], start_cell: args?.start_cell || 'A1', mode: args?.mode || 'create', }; const tmpSpec = resolved + `.spec_${Date.now()}.json`; try { fs.writeFileSync(tmpSpec, JSON.stringify(spec), 'utf-8'); const { stdout } = await execFileAsync( 'python3', ['-c', WRITE_SCRIPT, tmpSpec], { maxBuffer: 5 * 1024 * 1024, timeout: 30_000 } ); const result = JSON.parse(stdout.trim()); if (result.error) { if (result.error === 'openpyxl_missing') { return { success: false, error: 'openpyxl not installed. Run: pip3 install openpyxl' }; } return { success: false, error: result.error }; } const filename = path.basename(resolved); const rel = path.relative(workspacePath, resolved); return { success: true, stdout: `Excel file saved: ${filename} | Sheet: ${result.sheet} | Rows written: ${result.rows_written}`, data: { path: resolved, filename, sheet: result.sheet, rowsWritten: result.rows_written }, }; } catch (err: any) { return { success: false, error: `excel_write failed: ${err.message}` }; } finally { try { fs.unlinkSync(tmpSpec); } catch {} } }, };