""" skills/xlsx_skill.py - XLSX Excel 电子表格技能 ============================================= 基于 openpyxl 创建和编辑 Excel 文件，融入 MiniMax XLSX Skill 设计理念。 支持创建/编辑/格式化工作表、图表、公式等。 """ from __future__ import annotations import json import os import time from pathlib import Path from typing import Any, Dict, List, Optional from aiskills.base import Skill, SkillParameter, SkillResult class XLSXCreateSkill(Skill): """生成 Excel 电子表格 sheets 为 JSON 对象，key 为工作表名，value 为: { "headers": ["列1", "列2", ...], "rows": [["值1", "值2"], ...], "col_widths": [15, 20, ...], "freeze_panes": "A2", "auto_filter": true, "formulas": {"C2": "=A2+B2"}, } """ name = "xlsx_create" description = ( "生成 Excel (XLSX) 电子表格文件。支持多工作表、表头、数据行、公式、" "冻结窗格、自动筛选、列宽设置。" "【重要】必须提供 sheets 参数（JSON字符串），格式为：" '{"工作表名": {"headers":["列名1","列名2"], "rows":[["值1","值2"],["值3","值4"]]}}。' "参数 output_path 指定输出文件路径，默认为工作目录下。" ) category = "doc" dangerous = True parameters = [ SkillParameter("sheets", "string", "【必填】工作表数据（JSON字符串）。" "格式: '{\"Sheet1\":{\"headers\":[\"姓名\",\"年龄\"],\"rows\":[[\"Alice\",25],[\"Bob\",30]]}}'。" "headers=列名数组, rows=二维数组。可选: col_widths, formulas, freeze_panes, auto_filter", required=True), SkillParameter("output_path", "string", "输出文件路径(如 /tmp/report.xlsx)", required=False, default=""), SkillParameter("title", "string", "文档标题", required=False, default=""), ] async def execute(self, sheets: str = "", output_path: str = "", title: str = "", **kwargs) -> SkillResult: try: import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter except ImportError: return SkillResult(success=False, error="openpyxl 未安装: pip install openpyxl") # [v1.18.8] 容错: 如果 sheets 为空但 kwargs 中有 raw_input 或其他数据，尝试解析 if not sheets: raw = kwargs.get("raw_input", "") if raw: # 尝试直接将 raw_input 解析为 sheets JSON try: parsed = json.loads(raw) if isinstance(parsed, dict): sheets = raw elif isinstance(parsed, list) and parsed: # 如果是二维数组，自动包装为 Sheet1 sheets = json.dumps({"Sheet1": {"headers": [], "rows": parsed}}) except (json.JSONDecodeError, TypeError): pass # 如果仍然为空，尝试从 kwargs 中的 headers/rows 构建 if not sheets: headers = kwargs.get("headers") rows = kwargs.get("rows") if headers or rows: sheet_obj = {} if headers: sheet_obj["headers"] = headers if isinstance(headers, list) else json.loads(headers) if rows: sheet_obj["rows"] = rows if isinstance(rows, list) else json.loads(rows) if sheet_obj: sheets = json.dumps({"Sheet1": sheet_obj}) if not sheets: return SkillResult( success=False, error="缺少必需参数: sheets。请提供 JSON 格式的工作表数据，" '例如: {"Sheet1":{"headers":["姓名","年龄"],"rows":[["Alice",25]]}}' ) try: sheets_data = json.loads(sheets) if isinstance(sheets, str) else sheets if not isinstance(sheets_data, dict): return SkillResult(success=False, error="sheets 必须是 JSON 对象 (key=工作表名)") except json.JSONDecodeError as e: return SkillResult(success=False, error=f"sheets JSON 解析失败: {e}") try: if not output_path.strip(): default_dir = Path.home() / ".myagent" / "data" / "workspace" / "userfiles" default_dir.mkdir(parents=True, exist_ok=True) out = default_dir / f"report_{int(time.time())}.xlsx" else: out = Path(output_path).expanduser().resolve() out.parent.mkdir(parents=True, exist_ok=True) wb = openpyxl.Workbook() wb.remove(wb.active) header_font = Font(name="SimHei", size=11, bold=True, color="FFFFFF") header_fill = PatternFill(start_color="1A365D", end_color="1A365D", fill_type="solid") header_align = Alignment(horizontal="center", vertical="center", wrap_text=True) cell_font = Font(name="SimHei", size=10) cell_align = Alignment(vertical="center", wrap_text=True) formula_font = Font(name="Consolas", size=10, color="006400") thin_border = Border( left=Side(style="thin", color="D0D5DD"), right=Side(style="thin", color="D0D5DD"), top=Side(style="thin", color="D0D5DD"), bottom=Side(style="thin", color="D0D5DD"), ) alt_fill = PatternFill(start_color="F7FAFC", end_color="F7FAFC", fill_type="solid") total_rows = 0 for sheet_name, sheet_data in sheets_data.items(): if not isinstance(sheet_data, dict): continue ws = wb.create_sheet(title=sheet_name[:31]) headers = sheet_data.get("headers", []) rows = sheet_data.get("rows", []) col_widths = sheet_data.get("col_widths", []) formulas = sheet_data.get("formulas", {}) freeze_panes = sheet_data.get("freeze_panes", "") auto_filter = sheet_data.get("auto_filter", False) for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col_idx, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = header_align cell.border = thin_border for row_idx, row_data in enumerate(rows, 2): for col_idx, val in enumerate(row_data, 1): cell = ws.cell(row=row_idx, column=col_idx, value=val) cell.font = cell_font cell.alignment = cell_align cell.border = thin_border if row_idx % 2 == 0: cell.fill = alt_fill total_rows += 1 for cell_ref, formula in formulas.items(): cell = ws[cell_ref] cell.value = formula cell.font = formula_font cell.border = thin_border if not col_widths: for col_idx in range(1, len(headers) + 1): max_len = len(str(headers[col_idx - 1])) if col_idx <= len(headers) else 0 for row_data in rows[:50]: if col_idx <= len(row_data): max_len = max(max_len, len(str(row_data[col_idx - 1]))) ws.column_dimensions[get_column_letter(col_idx)].width = min(max(max_len + 4, 10), 50) else: for i, w in enumerate(col_widths, 1): ws.column_dimensions[get_column_letter(i)].width = w if freeze_panes: ws.freeze_panes = freeze_panes if auto_filter and headers: ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}{len(rows) + 1}" if title: wb.properties.title = title wb.save(str(out)) sheet_names = list(sheets_data.keys()) return SkillResult( success=True, message=f"Excel 文件已生成: {out} ({len(sheet_names)} 个工作表, {total_rows} 行数据)", files=[str(out)], data={"path": str(out), "sheets": sheet_names, "total_rows": total_rows}, ) except Exception as e: return SkillResult(success=False, error=f"XLSX 生成失败: {e}") class XLSXReadSkill(Skill): """读取 Excel 文件内容""" name = "xlsx_read" description = "读取 Excel (XLSX/XLS/CSV) 文件，提取数据。支持指定工作表和行数限制。" category = "doc" parameters = [ SkillParameter("path", "string", "Excel 文件路径", required=True), SkillParameter("sheet", "string", "工作表名 (空=第一个)", required=False, default=""), SkillParameter("max_rows", "integer", "最大读取行数", required=False, default=200), SkillParameter("max_chars", "integer", "最大字符数", required=False, default=50000), ] async def execute(self, path: str = "", sheet: str = "", max_rows: int = 200, max_chars: int = 50000, **kwargs) -> SkillResult: try: import openpyxl except ImportError: return SkillResult(success=False, error="openpyxl 未安装: pip install openpyxl") fp = Path(path).expanduser().resolve() if not fp.exists(): return SkillResult(success=False, error=f"文件不存在: {path}") try: wb = openpyxl.load_workbook(str(fp), read_only=True, data_only=True) if not wb.worksheets: wb.close() import csv text = fp.read_text(encoding="utf-8-sig", errors="replace") reader = csv.reader(text.splitlines()) rows = list(reader) if len(rows) > max_rows: rows = rows[:max_rows] full = "\n".join("\t".join(row) for row in rows) if len(full) > max_chars: full = full[:max_chars] + "\n..." return SkillResult(success=True, message=f"已读取 CSV: {fp.name} ({len(rows)} 行)", output=full, data={"rows": len(rows)}) sheet_names = wb.sheetnames ws_name = sheet if sheet and sheet in sheet_names else sheet_names[0] ws = wb[ws_name] parts = [] row_count = 0 for row in ws.iter_rows(values_only=True): if row_count >= max_rows: parts.append(f"\n... (仅显示前 {max_rows} 行)") break cells = [str(c) if c is not None else "" for c in row] parts.append("\t".join(cells)) row_count += 1 wb.close() full = "\n".join(parts) if len(full) > max_chars: full = full[:max_chars] + "\n..." return SkillResult( success=True, message=f"已读取 Excel: {fp.name} (工作表: {ws_name}, {row_count} 行)", data={"path": str(fp), "sheets": sheet_names, "active_sheet": ws_name, "rows": row_count}, output=full, ) except Exception as e: return SkillResult(success=False, error=f"XLSX 读取失败: {e}") class XLSXEditSkill(Skill): """编辑 Excel 文件""" name = "xlsx_edit" description = "编辑已有 Excel 文件。支持: append_row, add_column, update_cell。" category = "doc" dangerous = True parameters = [ SkillParameter("path", "string", "Excel 文件路径", required=True), SkillParameter("action", "string", "操作类型", required=True, enum=["append_row", "add_column", "update_cell"]), SkillParameter("sheet", "string", "工作表名 (空=第一个)", required=False, default=""), SkillParameter("data", "string", "操作数据 JSON。" "append_row: [[\"值1\",\"值2\"]]" "add_column: {\"header\":\"列名\", \"values\":[1,2,3]}" "update_cell: {\"cell\":\"A1\", \"value\":\"新值\"}", required=True), ] async def execute(self, path: str = "", action: str = "", sheet: str = "", data: str = "", **kwargs) -> SkillResult: try: import openpyxl from openpyxl.styles import Font, PatternFill from openpyxl.utils import get_column_letter except ImportError: return SkillResult(success=False, error="openpyxl 未安装") fp = Path(path).expanduser().resolve() if not fp.exists(): return SkillResult(success=False, error=f"文件不存在: {path}") try: data_parsed = json.loads(data) if isinstance(data, str) else data except json.JSONDecodeError as e: return SkillResult(success=False, error=f"data JSON 解析失败: {e}") try: wb = openpyxl.load_workbook(str(fp)) ws_name = sheet if sheet and sheet in wb.sheetnames else wb.sheetnames[0] ws = wb[ws_name] if action == "append_row": last_row = ws.max_row rows_to_add = data_parsed if isinstance(data_parsed[0], list) else [data_parsed] for row_data in rows_to_add: last_row += 1 for col_idx, val in enumerate(row_data, 1): ws.cell(row=last_row, column=col_idx, value=val) msg = f"追加 {len(rows_to_add)} 行到工作表 '{ws_name}'" elif action == "add_column": header = data_parsed.get("header", "New Column") values = data_parsed.get("values", []) formula = data_parsed.get("formula", "") new_col = ws.max_column + 1 ws.cell(row=1, column=new_col, value=header) ws.cell(row=1, column=new_col).font = Font(bold=True, color="FFFFFF") ws.cell(row=1, column=new_col).fill = PatternFill(start_color="1A365D", fill_type="solid") if formula: for i in range(2, ws.max_row + 1): cell = ws.cell(row=i, column=new_col) cell.value = formula.replace("{ROW}", str(i)) cell.font = Font(name="Consolas", color="006400") elif values: for i, val in enumerate(values): ws.cell(row=i + 2, column=new_col, value=val) msg = f"添加列 '{header}' 到工作表 '{ws_name}'" elif action == "update_cell": cell_ref = data_parsed.get("cell", "A1") value = data_parsed.get("value", "") ws[cell_ref] = value msg = f"更新单元格 {cell_ref} = {value}" else: return SkillResult(success=False, error=f"未知操作: {action}") wb.save(str(fp)) return SkillResult( success=True, message=f"编辑成功: {msg}", files=[str(fp)], data={"path": str(fp), "action": action, "sheet": ws_name}, ) except Exception as e: return SkillResult(success=False, error=f"XLSX 编辑失败: {e}")