Create Feishu sheets for spreadsheet files

backend/api/server.py

@@ -7,6 +7,7 @@ chat, slash commands, and saved sessions, and serves the static SPA.
 from __future__ import annotations
 
 import asyncio
+import csv
 import json
 import os
 import queue
@@ -58,9 +59,15 @@ API_TOOL_CONTENT_MAX_CHARS = 20000
 FEISHU_NPM_REGISTRY = 'https://pkgs.d.xiaomi.net/artifactory/api/npm/mi-npm/'
 FEISHU_MCP_PACKAGE = '@mi/feishu-mcp-pro@latest'
 FEISHU_MCP_SERVER_NAME = 'feishu-mcp-pro'
-FEISHU_SUPPORTED_DOCUMENT_SUFFIXES = {'.csv', '.docx', '.md', '.txt', '.xlsx'}
+FEISHU_DOCUMENT_SUFFIXES = {'.docx', '.md', '.txt'}
+FEISHU_SPREADSHEET_SUFFIXES = {'.csv', '.xlsx'}
+FEISHU_SUPPORTED_DOCUMENT_SUFFIXES = FEISHU_DOCUMENT_SUFFIXES | FEISHU_SPREADSHEET_SUFFIXES
 FEISHU_UNSUPPORTED_DOCUMENT_SUFFIXES = {'.json', '.jsonl'}
 FEISHU_DOC_MARKDOWN_MAX_CHARS = 100_000
+FEISHU_SPREADSHEET_MAX_ROWS = 5000
+FEISHU_SPREADSHEET_MAX_COLS = 200
+FEISHU_SPREADSHEET_MAX_CELL_CHARS = 5000
+FEISHU_SPREADSHEET_WRITE_BATCH_ROWS = 500
 FEISHU_ONLINE_DOCS_FILENAME = 'online-docs.json'
 
 
@@ -1076,29 +1083,37 @@ def create_app(state: AgentState) -> FastAPI:
                     'status': status,
                 },
             )
+        paths = _feishu_paths(state, payload.account_id)
+        runtime = MCPRuntime(servers=(_feishu_server_profile(paths),))
         title = _clean_feishu_doc_title(payload.title or file_path.stem)
+        kind = 'sheet' if suffix in FEISHU_SPREADSHEET_SUFFIXES else 'doc'
         try:
-            markdown = _convert_file_to_feishu_markdown(file_path, title=title)
+            if kind == 'sheet':
+                rendered, metadata = _create_feishu_online_spreadsheet(
+                    runtime,
+                    file_path,
+                    title=title,
+                    folder_token=payload.folder_token,
+                )
+            else:
+                markdown = _convert_file_to_feishu_markdown(file_path, title=title)
+                arguments: dict[str, Any] = {'title': title, 'markdown': markdown}
+                if payload.folder_token and payload.folder_token.strip():
+                    arguments['folder_token'] = payload.folder_token.strip()
+                rendered, metadata = runtime.call_tool(
+                    'doc_create',
+                    arguments=arguments,
+                    server_name=FEISHU_MCP_SERVER_NAME,
+                    max_chars=API_TOOL_CONTENT_MAX_CHARS,
+                    timeout_seconds=60.0,
+                )
         except DataAgentInputError as exc:
             raise HTTPException(status_code=400, detail=str(exc))
         except OSError as exc:
             raise HTTPException(status_code=400, detail=f'读取文件失败: {exc}')
-
-        arguments: dict[str, Any] = {'title': title, 'markdown': markdown}
-        if payload.folder_token and payload.folder_token.strip():
-            arguments['folder_token'] = payload.folder_token.strip()
-        paths = _feishu_paths(state, payload.account_id)
-        runtime = MCPRuntime(servers=(_feishu_server_profile(paths),))
-        try:
-            rendered, metadata = runtime.call_tool(
-                'doc_create',
-                arguments=arguments,
-                server_name=FEISHU_MCP_SERVER_NAME,
-                max_chars=API_TOOL_CONTENT_MAX_CHARS,
-                timeout_seconds=60.0,
-            )
         except Exception as exc:
-            raise HTTPException(status_code=502, detail=f'飞书文档创建失败: {exc}')
+            target_name = '飞书表格' if kind == 'sheet' else '飞书文档'
+            raise HTTPException(status_code=502, detail=f'{target_name}创建失败: {exc}')
         url = _extract_first_url(rendered)
         if url:
             _record_feishu_online_doc(
@@ -1107,11 +1122,13 @@ def create_app(state: AgentState) -> FastAPI:
                 file_path=file_path,
                 title=title,
                 url=url,
+                kind=kind,
             )
         return {
             'ok': True,
             'title': title,
             'url': url,
+            'kind': kind,
             'file_path': str(file_path),
             'file_type': suffix.lstrip('.'),
             'result': rendered,
@@ -2553,6 +2570,7 @@ def _record_feishu_online_doc(
     file_path: Path,
     title: str,
     url: str,
+    kind: str = 'doc',
 ) -> None:
     paths = _feishu_paths(state, account_id)
     map_path = paths['root'] / FEISHU_ONLINE_DOCS_FILENAME
@@ -2575,6 +2593,7 @@ def _record_feishu_online_doc(
         files[str(file_path)] = {
             'url': clean_url,
             'title': title,
+            'kind': kind,
             'file_path': str(file_path),
             'created_at': created_at,
             'updated_at': now,
@@ -2604,6 +2623,290 @@ def _clean_feishu_doc_title(value: str) -> str:
     return title[:80].strip() or 'Claw 生成文档'
 
 
+def _create_feishu_online_spreadsheet(
+    runtime: MCPRuntime,
+    file_path: Path,
+    *,
+    title: str,
+    folder_token: str | None = None,
+) -> tuple[str, dict[str, Any]]:
+    sheets = _load_file_as_feishu_sheets(file_path)
+    create_params: dict[str, Any] = {'title': title}
+    if folder_token and folder_token.strip():
+        create_params['folder_token'] = folder_token.strip()
+    rendered, metadata = runtime.call_tool(
+        'sheet_ops',
+        arguments={'action': 'create', 'params': create_params},
+        server_name=FEISHU_MCP_SERVER_NAME,
+        max_chars=API_TOOL_CONTENT_MAX_CHARS,
+        timeout_seconds=60.0,
+    )
+    spreadsheet_token = _extract_feishu_spreadsheet_token(rendered)
+    if not spreadsheet_token:
+        raise DataAgentInputError('飞书表格已创建，但没有解析到 spreadsheet token')
+
+    sheet_refs = _discover_feishu_sheet_refs(runtime, spreadsheet_token)
+    for index, sheet_data in enumerate(sheets):
+        rows = sheet_data['rows']
+        if not rows:
+            continue
+        if index == 0:
+            sheet_ref = sheet_refs[0] if sheet_refs else 'Sheet1'
+        else:
+            sheet_ref = _create_feishu_worksheet(
+                runtime,
+                spreadsheet_token,
+                title=str(sheet_data.get('title') or f'Sheet{index + 1}'),
+                index=index,
+            )
+        _write_feishu_sheet_rows(runtime, spreadsheet_token, sheet_ref, rows)
+    return rendered, metadata
+
+
+def _load_file_as_feishu_sheets(file_path: Path) -> list[dict[str, Any]]:
+    suffix = file_path.suffix.lower()
+    if suffix == '.csv':
+        rows = _load_csv_rows_for_feishu(file_path)
+        return [{'title': _clean_feishu_sheet_title(file_path.stem) or 'Sheet1', 'rows': rows}]
+    if suffix == '.xlsx':
+        return _load_xlsx_rows_for_feishu(file_path)
+    raise DataAgentInputError(f'不支持转为飞书表格的文件类型: {suffix}')
+
+
+def _load_csv_rows_for_feishu(file_path: Path) -> list[list[str]]:
+    rows: list[list[str]] = []
+    with file_path.open('r', encoding='utf-8-sig', errors='replace', newline='') as handle:
+        reader = csv.reader(handle)
+        for row_index, row in enumerate(reader):
+            if row_index >= FEISHU_SPREADSHEET_MAX_ROWS:
+                break
+            rows.append(_clean_feishu_sheet_row(row))
+    normalized_rows = _normalize_feishu_sheet_rows(rows)
+    if not normalized_rows:
+        raise DataAgentInputError('没有解析到可转成飞书表格的数据')
+    return normalized_rows
+
+
+def _load_xlsx_rows_for_feishu(file_path: Path) -> list[dict[str, Any]]:
+    try:
+        import openpyxl  # type: ignore[import-not-found]
+    except ImportError as exc:
+        raise DataAgentInputError('解析 xlsx 需要 openpyxl') from exc
+    workbook = openpyxl.load_workbook(file_path, data_only=True, read_only=True)
+    sheets: list[dict[str, Any]] = []
+    for index, sheet in enumerate(workbook.worksheets):
+        if hasattr(sheet, 'reset_dimensions'):
+            sheet.reset_dimensions()
+        rows: list[list[str]] = []
+        for row_index, row in enumerate(sheet.iter_rows(values_only=True)):
+            if row_index >= FEISHU_SPREADSHEET_MAX_ROWS:
+                break
+            rows.append(_clean_feishu_sheet_row(row))
+        normalized_rows = _normalize_feishu_sheet_rows(rows)
+        if normalized_rows:
+            sheets.append(
+                {
+                    'title': _clean_feishu_sheet_title(str(sheet.title)) or f'Sheet{index + 1}',
+                    'rows': normalized_rows,
+                }
+            )
+    if not sheets:
+        raise DataAgentInputError('没有解析到可转成飞书表格的数据')
+    return sheets
+
+
+def _clean_feishu_sheet_row(row: Any) -> list[str]:
+    if not isinstance(row, (list, tuple)):
+        return []
+    return [_clean_feishu_sheet_cell(cell) for cell in row[:FEISHU_SPREADSHEET_MAX_COLS]]
+
+
+def _clean_feishu_sheet_cell(value: Any) -> str:
+    if value is None:
+        return ''
+    text = str(value)
+    if len(text) > FEISHU_SPREADSHEET_MAX_CELL_CHARS:
+        return text[:FEISHU_SPREADSHEET_MAX_CELL_CHARS]
+    return text
+
+
+def _normalize_feishu_sheet_rows(rows: list[list[str]]) -> list[list[str]]:
+    while rows and not any(cell for cell in rows[-1]):
+        rows.pop()
+    if not rows:
+        return []
+    width = max((len(row) for row in rows), default=0)
+    width = min(max(width, 1), FEISHU_SPREADSHEET_MAX_COLS)
+    return [row[:width] + [''] * max(0, width - len(row)) for row in rows]
+
+
+def _clean_feishu_sheet_title(value: str) -> str:
+    title = re.sub(r'[\r\n\t/\\?*\[\]:]+', ' ', value).strip()
+    return title[:80].strip()
+
+
+def _extract_feishu_spreadsheet_token(text: str) -> str | None:
+    url = _extract_first_url(text)
+    candidates = [item for item in (url, text) if item]
+    for candidate in candidates:
+        match = re.search(r'/sheets/([A-Za-z0-9]+)', candidate)
+        if match:
+            return match.group(1)
+    payload = _parse_first_json_object(text)
+    token = _find_first_string_value(
+        payload,
+        {'spreadsheet_token', 'spreadsheetToken', 'token'},
+    )
+    return token.strip() if token else None
+
+
+def _discover_feishu_sheet_refs(runtime: MCPRuntime, spreadsheet_token: str) -> list[str]:
+    try:
+        rendered, _metadata = runtime.call_tool(
+            'sheet_ops',
+            arguments={'action': 'meta', 'params': {'url_or_token': spreadsheet_token}},
+            server_name=FEISHU_MCP_SERVER_NAME,
+            max_chars=API_TOOL_CONTENT_MAX_CHARS,
+            timeout_seconds=60.0,
+        )
+    except Exception:
+        return []
+    refs = _extract_feishu_sheet_refs(rendered)
+    return refs
+
+
+def _extract_feishu_sheet_refs(text: str) -> list[str]:
+    payload = _parse_first_json_object(text)
+    refs: list[str] = []
+
+    def visit(value: Any) -> None:
+        if isinstance(value, dict):
+            raw_id = value.get('sheet_id') or value.get('sheetId')
+            raw_title = value.get('title') or value.get('name')
+            if isinstance(raw_id, str) and raw_id.strip():
+                refs.append(raw_id.strip())
+            elif isinstance(raw_title, str) and raw_title.strip():
+                refs.append(raw_title.strip())
+            for child in value.values():
+                visit(child)
+        elif isinstance(value, list):
+            for child in value:
+                visit(child)
+
+    visit(payload)
+    return _dedupe_strings([ref for ref in refs if ref])
+
+
+def _dedupe_strings(values: list[str]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for value in values:
+        if value in seen:
+            continue
+        seen.add(value)
+        result.append(value)
+    return result
+
+
+def _create_feishu_worksheet(
+    runtime: MCPRuntime,
+    spreadsheet_token: str,
+    *,
+    title: str,
+    index: int,
+) -> str:
+    sheet_title = _clean_feishu_sheet_title(title) or f'Sheet{index + 1}'
+    rendered, _metadata = runtime.call_tool(
+        'sheet_ops',
+        arguments={
+            'action': 'add_sheet',
+            'params': {
+                'spreadsheet_token': spreadsheet_token,
+                'title': sheet_title,
+                'index': index,
+            },
+        },
+        server_name=FEISHU_MCP_SERVER_NAME,
+        max_chars=API_TOOL_CONTENT_MAX_CHARS,
+        timeout_seconds=60.0,
+    )
+    refs = _extract_feishu_sheet_refs(rendered)
+    return refs[0] if refs else sheet_title
+
+
+def _write_feishu_sheet_rows(
+    runtime: MCPRuntime,
+    spreadsheet_token: str,
+    sheet_ref: str,
+    rows: list[list[str]],
+) -> None:
+    if not rows:
+        return
+    width = max(len(row) for row in rows)
+    end_column = _spreadsheet_column_name(width)
+    for start in range(0, len(rows), FEISHU_SPREADSHEET_WRITE_BATCH_ROWS):
+        chunk = rows[start : start + FEISHU_SPREADSHEET_WRITE_BATCH_ROWS]
+        start_row = start + 1
+        end_row = start + len(chunk)
+        runtime.call_tool(
+            'sheet_ops',
+            arguments={
+                'action': 'write',
+                'params': {
+                    'spreadsheet_token': spreadsheet_token,
+                    'range': f'{sheet_ref}!A{start_row}:{end_column}{end_row}',
+                    'values': json.dumps(chunk, ensure_ascii=False),
+                },
+            },
+            server_name=FEISHU_MCP_SERVER_NAME,
+            max_chars=API_TOOL_CONTENT_MAX_CHARS,
+            timeout_seconds=60.0,
+        )
+
+
+def _spreadsheet_column_name(index: int) -> str:
+    if index <= 0:
+        raise ValueError('index must be positive')
+    name = ''
+    current = index
+    while current:
+        current, remainder = divmod(current - 1, 26)
+        name = chr(ord('A') + remainder) + name
+    return name
+
+
+def _parse_first_json_object(text: str) -> Any:
+    stripped = text.strip()
+    candidates = [stripped]
+    json_match = re.search(r'(\{.*\})', stripped, flags=re.DOTALL)
+    if json_match:
+        candidates.append(json_match.group(1))
+    for candidate in candidates:
+        try:
+            return json.loads(candidate)
+        except json.JSONDecodeError:
+            continue
+    return None
+
+
+def _find_first_string_value(value: Any, keys: set[str]) -> str | None:
+    if isinstance(value, dict):
+        for key in keys:
+            item = value.get(key)
+            if isinstance(item, str) and item.strip():
+                return item
+        for child in value.values():
+            found = _find_first_string_value(child, keys)
+            if found:
+                return found
+    elif isinstance(value, list):
+        for child in value:
+            found = _find_first_string_value(child, keys)
+            if found:
+                return found
+    return None
+
+
 def _convert_file_to_feishu_markdown(file_path: Path, *, title: str) -> str:
     suffix = file_path.suffix.lower()
     if suffix == '.md':