Dingtalk Ai Table

Bud1

GELOG.

 
@
�





 
@
�






 
@
�






 
@
CHANGELOG.mdIlocblobB.������package.jsonIlocblob�.������	README.mdIlocblob
".������
referencesIlocblob
�.������scriptsIlocblob.������scriptsbwspblob�bplist00�
]ShowStatusBar[ShowToolbar[ShowTabView_ContainerShowSidebar\WindowBounds[ShowSidebar		_{{159, 221}, {1298, 749}}	#/;R_klmno�


�scriptsicvpblob
�bplist00�
	

_backgroundColorBlue[gridSpacingXtextSize_backgroundColorRed^backgroundType_backgroundColorGreen[gridOffsetX[gridOffsetY_scrollPositionY\showItemInfo_viewOptionsVersion_scrollPositionXYarrangeBy]labelOnBottom_showIconPreviewXiconSize#?�#@K#@(#
#�gTnone		#@P+AMVkz��������


"
+
4
=
?
H
I
Z
c
h
i
j

sscriptsvSrnlong
SKILL.mdIlocblobr.������testsIlocblob�.������testsbwspblob�bplist00�
]ShowStatusBar[ShowToolbar[ShowTabView_ContainerShowSidebar\WindowBounds[ShowSidebar		_{{159, 221}, {1298, 749}}	#/;R_klmno�


�testsicvpblob
�bplist00�
	

_backgroundColorBlue[gridSpacingXtextSize_backgroundColorRed^backgroundType_backgroundColorGreen[gridOffsetX[gridOffsetY_scrollPositionY\showItemInfo_viewOptionsVersion_scrollPositionXYarrangeBy]labelOnBottom_showIconPreviewXiconSize#?�#@K#@(#
#�gTnone		#@P+AMVkz��������


"
+
4
=
?
H
I
Z
c
h
i
j

stestsvSrnlong
E
DSDB
 `
�





 
@
�






 
@
�






 
@


�testsicvpblob
�bplist00�
	

_backgroundColorBlue[gridSpacingXtextSize_backgroundColorRed^backgroundType_backgroundColorGreen[gridOffsetX[gridOffsetY_scrollPositionY\showItemInfo_viewOptionsVersion_scrollPositionXYarrangeBy]labelOnBottom_showIconPreviewXiconSize#?�#@K#@(#
#�gTnone		#@P+AMVkz��������


"
+
4
=
?
H
I
Z
c
h
i
j

stestsvSrnlong

#!/bin/bash
# 示例 1：列出所有可访问的 Base

MCP_URL="${DINGTALK_MCP_URL}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

echo "📋 列出所有 Base..."
mcporter call "$MCP_URL" .list_bases limit=10

#!/bin/bash
# 示例 2：创建新 Base

MCP_URL="${DINGTALK_MCP_URL}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

BASE_NAME="${1:-我的项目}"

echo "🆕 创建 Base: $BASE_NAME"
mcporter call "$MCP_URL" .create_base baseName="$BASE_NAME"

#!/bin/bash
# 示例 3：查看 Base 内的表

MCP_URL="${DINGTALK_MCP_URL}"
BASE_ID="${1}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

if [ -z "$BASE_ID" ]; then
  echo "❌ 用法：$0 <baseId>"
  exit 1
fi

echo "📊 查看 Base 内的表..."
mcporter call "$MCP_URL" .get_base baseId="$BASE_ID"

#!/bin/bash
# 示例 4：查询记录

MCP_URL="${DINGTALK_MCP_URL}"
BASE_ID="${1}"
TABLE_ID="${2}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

if [ -z "$BASE_ID" ] || [ -z "$TABLE_ID" ]; then
  echo "❌ 用法：$0 <baseId> <tableId>"
  exit 1
fi

echo "🔍 查询记录..."
mcporter call "$MCP_URL" .query_records \
  --args "{\"baseId\":\"$BASE_ID\",\"tableId\":\"$TABLE_ID\",\"limit\":10}"

#!/bin/bash
# 示例 5：新增记录

MCP_URL="${DINGTALK_MCP_URL}"
BASE_ID="${1}"
TABLE_ID="${2}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

if [ -z "$BASE_ID" ] || [ -z "$TABLE_ID" ]; then
  echo "❌ 用法：$0 <baseId> <tableId>"
  exit 1
fi

echo "➕ 新增记录..."
mcporter call "$MCP_URL" .create_records \
  --args "{
    \"baseId\":\"$BASE_ID\",
    \"tableId\":\"$TABLE_ID\",
    \"records\":[
      {\"cells\":{\"fld_name\":\"张三\",\"fld_age\":25}},
      {\"cells\":{\"fld_name\":\"李四\",\"fld_age\":30}}
    ]
  }"

#!/bin/bash
# 示例 6：批量导入记录

MCP_URL="${DINGTALK_MCP_URL}"
BASE_ID="${1}"
TABLE_ID="${2}"
CSV_FILE="${3}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

if [ -z "$BASE_ID" ] || [ -z "$TABLE_ID" ] || [ -z "$CSV_FILE" ]; then
  echo "❌ 用法：$0 <baseId> <tableId> <csv_file>"
  exit 1
fi

echo "📥 批量导入记录..."
python3 scripts/import_records.py "$BASE_ID" "$TABLE_ID" "$CSV_FILE"

#!/bin/bash
# 示例 7：批量新增字段

MCP_URL="${DINGTALK_MCP_URL}"
BASE_ID="${1}"
TABLE_ID="${2}"
FIELDS_FILE="${3}"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

if [ -z "$BASE_ID" ] || [ -z "$TABLE_ID" ] || [ -z "$FIELDS_FILE" ]; then
  echo "❌ 用法：$0 <baseId> <tableId> <fields_file>"
  exit 1
fi

echo "🆕 批量新增字段..."
python3 scripts/bulk_add_fields.py "$BASE_ID" "$TABLE_ID" "$FIELDS_FILE"

{
  "name": "dingtalk-ai-table",
  "version": "0.6.0",
  "description": "钉钉 AI 表格（多维表）操作技能。基于新版 MCP tools，使用 baseId / tableId / fieldId / recordId 体系执行 Base、Table、Field、Record 管理。脚本文件 I/O 限制在工作区内。",
  "keywords": [
    "dingtalk",
    "ai-table",
    "mcp",
    "多维表",
    "openclaw",
    "skill"
  ],
  "author": "Marila@Dingtalk",
  "contributors": [
    "Marila@Dingtalk"
  ],
  "license": "MIT",
  "homepage": "https://clawhub.com/skills/dingtalk-ai-table",
  "repository": {
    "type": "git",
    "url": "https://github.com/aliramw/dingtalk-ai-table.git"
  },
  "bugs": {
    "url": "https://github.com/aliramw/dingtalk-ai-table/issues"
  },
  "engines": {
    "node": ">=18.0.0"
  },
  "peerDependencies": {
    "mcporter": ">=0.8.1"
  },
  "clawhub": {
    "requiresBinaries": [
      "mcporter",
      "python3"
    ],
    "requiredEnv": [
      "DINGTALK_MCP_URL",
      "OPENCLAW_WORKSPACE"
    ],
    "credentials": [
      {
        "name": "DINGTALK_MCP_URL",
        "description": "钉钉 MCP Server Streamable HTTP URL (含访问令牌)",
        "docs": "https://mcp.dingtalk.com/#/detail?mcpId=9555",
        "storageMethod": "mcporter config (recommended) or environment variable"
      },
      {
        "name": "OPENCLAW_WORKSPACE",
        "description": "本地脚本文件读写沙箱根目录；建议设置为 ~/.openclaw/workspace",
        "docs": "https://github.com/aliramw/dingtalk-ai-table",
        "storageMethod": "environment variable"
      }
    ]
  },
  "scripts": {
    "test": "python3 tests/test_security.py && python3 tests/test_triggering.py"
  }
}

#!/usr/bin/env python3
"""
批量添加字段到钉钉 AI 表格数据表（新 MCP schema）

用法:
    python bulk_add_fields.py <baseId> <tableId> fields.json

fields.json 格式:
    [
        {"fieldName": "字段 1", "type": "text"},
        {"fieldName": "字段 2", "type": "number", "config": {"formatter": "INT"}},
        {"fieldName": "字段 3", "type": "singleSelect", "config": {"options": [{"name": "高"}]}}
    ]

兼容写法：
- name 会自动映射为 fieldName
- phone 会自动映射为 telephone
"""

import sys
import json
import subprocess
import os
import re
from functools import lru_cache
from pathlib import Path
from typing import Union, List, Dict, Any, Optional, Tuple

JsonData = Union[List[Any], Dict[str, Any]]

MAX_FILE_SIZE = 10 * 1024 * 1024
ALLOWED_FILE_EXTENSIONS = ['.json']
RESOURCE_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]{8,128}$')
ALLOWED_FIELD_TYPES = {
    'text', 'number', 'singleSelect', 'multipleSelect', 'date', 'currency',
    'user', 'department', 'group', 'progress', 'rating', 'checkbox',
    'attachment', 'url', 'richText', 'telephone', 'email', 'idCard',
    'barcode', 'geolocation', 'primaryDoc', 'formula', 'unidirectionalLink',
    'bidirectionalLink', 'creator', 'lastModifier', 'createdTime', 'lastModifiedTime'
}
FIELD_TYPE_ALIASES = {
    'phone': 'telephone',
}
MCPORTER_VERSION_PATTERN = re.compile(r'(\d+)\.(\d+)\.(\d+)')
MCPORTER_TEXT_OUTPUT_CUTOFF = (0, 8, 1)


def resolve_safe_path(path: str, allowed_root: Optional[str] = None) -> Path:
    if allowed_root is None:
        allowed_root = os.environ.get('OPENCLAW_WORKSPACE', os.getcwd())

    allowed_root = Path(allowed_root).resolve()
    target_path = Path(path).resolve() if Path(path).is_absolute() else (Path.cwd() / path).resolve()

    try:
        target_path.relative_to(allowed_root)
        return target_path
    except ValueError:
        raise ValueError(
            f"路径超出允许范围：{path}\n"
            f"目标路径：{target_path}\n"
            f"允许根目录：{allowed_root}\n"
            f"提示：设置 OPENCLAW_WORKSPACE 环境变量或确保文件在工作目录内"
        )


def validate_resource_id(resource_id: str) -> bool:
    return bool(resource_id and RESOURCE_ID_PATTERN.match(resource_id.strip()))


def validate_dentry_uuid(dentry_uuid: str) -> bool:
    """兼容旧测试名；新 schema 实际用于校验 baseId/tableId/fieldId。"""
    return validate_resource_id(dentry_uuid)


def validate_file_extension(filename: str, allowed_extensions: list) -> bool:
    return any(filename.lower().endswith(ext) for ext in allowed_extensions)


def parse_mcporter_version(raw_text: str) -> Optional[Tuple[int, int, int]]:
    match = MCPORTER_VERSION_PATTERN.search(raw_text)
    if not match:
        return None
    return tuple(int(part) for part in match.groups())


@lru_cache(maxsize=1)
def get_mcporter_version() -> Optional[Tuple[int, int, int]]:
    for cmd in (['mcporter', '--version'], ['mcporter', 'version']):
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        except (subprocess.TimeoutExpired, FileNotFoundError):
            return None

        if result.returncode != 0:
            continue

        version = parse_mcporter_version(f"{result.stdout}\n{result.stderr}")
        if version is not None:
            return version

    return None


def build_mcporter_call(args: List[str]) -> List[str]:
    cmd = ['mcporter', 'call', 'dingtalk-ai-table']
    version = get_mcporter_version()
    if version is not None and version < MCPORTER_TEXT_OUTPUT_CUTOFF:
        cmd.extend(['--output', 'text'])
    return cmd + args


def safe_json_load(file_path: Path, max_size: int = MAX_FILE_SIZE) -> JsonData:
    file_size = file_path.stat().st_size
    if file_size > max_size:
        raise ValueError(f"文件过大：{file_size:,} 字节 (限制：{max_size:,} 字节)")
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)


def normalize_field_config(field: Dict[str, Any]) -> Dict[str, Any]:
    normalized = dict(field)
    if 'fieldName' not in normalized and 'name' in normalized:
        normalized['fieldName'] = normalized.pop('name')
    normalized['type'] = FIELD_TYPE_ALIASES.get(normalized.get('type', 'text'), normalized.get('type', 'text'))
    return normalized


def validate_field_config(field: Dict[str, Any]) -> Tuple[bool, str]:
    if not isinstance(field, dict):
        return False, '字段配置必须是对象'

    field = normalize_field_config(field)

    if 'fieldName' not in field:
        return False, '缺少必需字段：fieldName'
    if not isinstance(field['fieldName'], str) or not field['fieldName'].strip():
        return False, 'fieldName 必须是非空字符串'

    field_type = field.get('type', 'text')
    if field_type not in ALLOWED_FIELD_TYPES:
        return False, f"不支持的字段类型：{field_type}"

    config = field.get('config')
    if config is not None and not isinstance(config, dict):
        return False, 'config 必须是对象'

    if field_type in {'singleSelect', 'multipleSelect'}:
        options = (config or {}).get('options')
        if not options or not isinstance(options, list):
            return False, 'singleSelect / multipleSelect 必须提供 config.options 数组'

    if field_type in {'unidirectionalLink', 'bidirectionalLink'}:
        linked_sheet_id = (config or {}).get('linkedSheetId')
        if not linked_sheet_id or not validate_resource_id(linked_sheet_id):
            return False, '关联字段必须提供合法的 config.linkedSheetId'

    return True, ''


def build_create_fields_payload(base_id: str, table_id: str, fields: List[Dict[str, Any]]) -> Dict[str, Any]:
    payload_fields = []
    for field in fields:
        normalized = normalize_field_config(field)
        item = {
            'fieldName': normalized['fieldName'].strip(),
            'type': normalized.get('type', 'text')
        }
        if 'config' in normalized and normalized['config'] is not None:
            item['config'] = normalized['config']
        payload_fields.append(item)
    return {
        'baseId': base_id,
        'tableId': table_id,
        'fields': payload_fields,
    }


def run_mcporter(args: List[str]) -> Optional[Dict[str, Any]]:
    if not args:
        print('错误：空命令')
        return None

    cmd = build_mcporter_call(args)
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        if result.returncode != 0:
            print(f"错误：{result.stderr.strip()}")
            return None
        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError as e:
            print(f"无法解析响应：{result.stdout[:200]}...")
            print(f"JSON 解析错误：{e}")
            return None
    except subprocess.TimeoutExpired:
        print('错误：命令执行超时（60 秒）')
        return None
    except FileNotFoundError:
        print('错误：未找到 mcporter 命令，请确认已安装')
        return None


def bulk_add_fields(base_id: str, table_id: str, fields_file: str) -> bool:
    try:
        safe_path = resolve_safe_path(fields_file)
    except ValueError as e:
        print(f"路径验证失败：{e}")
        return False

    if not validate_file_extension(fields_file, ALLOWED_FILE_EXTENSIONS):
        print(f"错误：只允许 {', '.join(ALLOWED_FILE_EXTENSIONS)} 文件")
        return False
    if not safe_path.exists():
        print(f"错误：文件不存在：{safe_path}")
        return False

    try:
        fields = safe_json_load(safe_path)
    except ValueError as e:
        print(f"错误：{e}")
        return False
    except json.JSONDecodeError as e:
        print(f"错误：JSON 格式无效：{e}")
        return False

    if not isinstance(fields, list) or not fields:
        print('错误：fields.json 必须是非空 JSON 数组')
        return False
    if len(fields) > 15:
        print('错误：单次最多创建 15 个字段，请拆分后重试')
        return False

    for i, field in enumerate(fields):
        valid, error = validate_field_config(field)
        if not valid:
            print(f"错误：字段 #{i+1} 配置无效：{error}")
            return False

    payload = build_create_fields_payload(base_id, table_id, fields)
    result = run_mcporter(['create_fields', '--args', json.dumps(payload, ensure_ascii=False)])

    if not result:
        return False

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return True


def main():
    if len(sys.argv) != 4:
        print(__doc__)
        print('用法示例:')
        print('  python bulk_add_fields.py basexxx tablexxx fields.json')
        sys.exit(1)

    base_id = sys.argv[1]
    table_id = sys.argv[2]
    fields_file = sys.argv[3]

    if not validate_resource_id(base_id):
        print('错误：无效的 baseId 格式')
        sys.exit(1)
    if not validate_resource_id(table_id):
        print('错误：无效的 tableId 格式')
        sys.exit(1)

    success = bulk_add_fields(base_id, table_id, fields_file)
    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()

#!/bin/bash
# 自动检查 dingtalk-ai-table MCP schema 版本
# 一次性检查策略：同一 MCP Server 地址只检查一次

set -e

MCP_URL="${DINGTALK_MCP_URL:-}"
WORKSPACE="${OPENCLAW_WORKSPACE:-$HOME/.openclaw/workspace}"
CACHE_DIR="$WORKSPACE/.cache/dingtalk-ai-table"

if [ -z "$MCP_URL" ]; then
  echo "❌ 错误：未设置 DINGTALK_MCP_URL"
  exit 1
fi

# 生成 URL hash 作为检查标记
URL_HASH=$(echo -n "$MCP_URL" | md5sum | cut -d' ' -f1)
CACHE_FILE="$CACHE_DIR/schema-check-$URL_HASH.json"

# 如果已检查过且结果为新版，直接跳过
if [ -f "$CACHE_FILE" ]; then
  RESULT=$(cat "$CACHE_FILE" | grep -o '"status":"[^"]*"' | cut -d'"' -f4)
  if [ "$RESULT" = "new_schema" ]; then
    echo "✅ 已确认新版 schema（缓存）"
    exit 0
  fi
fi

# 执行检查
echo "🔍 检查 MCP schema 版本..."
SCHEMA=$(mcporter list dingtalk-ai-table --schema 2>/dev/null || echo "")

if echo "$SCHEMA" | grep -q "list_bases\|get_base\|create_records"; then
  echo "✅ 确认新版 schema"
  mkdir -p "$CACHE_DIR"
  echo "{\"status\":\"new_schema\",\"checked_at\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"}" > "$CACHE_FILE"
  exit 0
else
  echo "❌ 检测到旧版 schema"
  echo ""
  echo "请按以下步骤更新："
  echo "1. 打开：https://mcp.dingtalk.com/#/detail?mcpId=9555&detailType=marketMcpDetail"
  echo "2. 点击右侧「获取 MCP Server 配置」"
  echo "3. 复制新的 MCP Server 地址"
  echo "4. 运行：mcporter config update dingtalk-ai-table --url '<新地址>'"
  echo "5. 重新运行此脚本"
  exit 1
fi

#!/usr/bin/env python3
"""
从 CSV / JSON 批量导入记录到钉钉 AI 表格（新 MCP schema）

用法:
    python import_records.py <baseId> <tableId> data.csv [batch_size]
    python import_records.py <baseId> <tableId> data.json [batch_size]

说明：
- CSV 表头默认视为 fieldId
- JSON 支持两种格式：
  1. [{"cells": {"fldxxx": "value"}}, ...]
  2. [{"fldxxx": "value"}, ...]  # 会自动包装成 cells
"""

import sys
import csv
import json
import subprocess
import os
import re
from functools import lru_cache
from pathlib import Path
from typing import Union, List, Dict, Any, Optional, Tuple

JsonData = Union[List[Any], Dict[str, Any]]
RecordDict = Dict[str, str]

MAX_FILE_SIZE = 50 * 1024 * 1024
ALLOWED_CSV_EXTENSIONS = ['.csv']
ALLOWED_JSON_EXTENSIONS = ['.json']
RESOURCE_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]{8,128}$')
MAX_RECORDS_PER_BATCH = 100
DEFAULT_BATCH_SIZE = 50
MCPORTER_VERSION_PATTERN = re.compile(r'(\d+)\.(\d+)\.(\d+)')
MCPORTER_TEXT_OUTPUT_CUTOFF = (0, 8, 1)


def resolve_safe_path(path: str, allowed_root: Optional[str] = None) -> Path:
    if allowed_root is None:
        allowed_root = os.environ.get('OPENCLAW_WORKSPACE', os.getcwd())
    allowed_root = Path(allowed_root).resolve()
    target_path = Path(path).resolve() if Path(path).is_absolute() else (Path.cwd() / path).resolve()
    try:
        target_path.relative_to(allowed_root)
        return target_path
    except ValueError:
        raise ValueError(
            f"路径超出允许范围：{path}\n"
            f"目标路径：{target_path}\n"
            f"允许根目录：{allowed_root}\n"
            f"提示：设置 OPENCLAW_WORKSPACE 环境变量或确保文件在工作目录内"
        )


def validate_resource_id(resource_id: str) -> bool:
    return bool(resource_id and RESOURCE_ID_PATTERN.match(resource_id.strip()))


def validate_dentry_uuid(dentry_uuid: str) -> bool:
    return validate_resource_id(dentry_uuid)


def validate_file_extension(filename: str, allowed_extensions: list) -> bool:
    return any(filename.lower().endswith(ext) for ext in allowed_extensions)


def parse_mcporter_version(raw_text: str) -> Optional[Tuple[int, int, int]]:
    match = MCPORTER_VERSION_PATTERN.search(raw_text)
    if not match:
        return None
    return tuple(int(part) for part in match.groups())


@lru_cache(maxsize=1)
def get_mcporter_version() -> Optional[Tuple[int, int, int]]:
    for cmd in (['mcporter', '--version'], ['mcporter', 'version']):
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        except (subprocess.TimeoutExpired, FileNotFoundError):
            return None

        if result.returncode != 0:
            continue

        version = parse_mcporter_version(f"{result.stdout}\n{result.stderr}")
        if version is not None:
            return version

    return None


def build_mcporter_call(args: List[str]) -> List[str]:
    cmd = ['mcporter', 'call', 'dingtalk-ai-table']
    version = get_mcporter_version()
    if version is not None and version < MCPORTER_TEXT_OUTPUT_CUTOFF:
        cmd.extend(['--output', 'text'])
    return cmd + args


def safe_csv_load(file_path: Path, max_size: int = MAX_FILE_SIZE) -> List[RecordDict]:
    file_size = file_path.stat().st_size
    if file_size > max_size:
        raise ValueError(f"文件过大：{file_size:,} 字节 (限制：{max_size:,} 字节)")
    with open(file_path, 'r', encoding='utf-8', newline='') as f:
        return list(csv.DictReader(f))


def safe_json_load(file_path: Path, max_size: int = MAX_FILE_SIZE) -> JsonData:
    file_size = file_path.stat().st_size
    if file_size > max_size:
        raise ValueError(f"文件过大：{file_size:,} 字节 (限制：{max_size:,} 字节)")
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)


def sanitize_record_value(value: Any) -> Optional[Union[str, int, float, bool, list, dict]]:
    if value is None:
        return None
    if isinstance(value, (bool, int, float, list, dict)):
        return value
    if not isinstance(value, str):
        return value
    if not value.strip():
        return None

    value = value.strip()
    if value.lower() == 'true':
        return True
    if value.lower() == 'false':
        return False

    try:
        if '.' in value:
            return float(value)
        return int(value)
    except ValueError:
        return value


def normalize_record(record: Dict[str, Any]) -> Dict[str, Any]:
    if 'cells' in record and isinstance(record['cells'], dict):
        cells = record['cells']
    else:
        cells = record
    normalized = {}
    for key, value in cells.items():
        sanitized = sanitize_record_value(value)
        if sanitized is not None:
            normalized[key] = sanitized
    return {'cells': normalized}


def validate_record(record: Dict[str, Any], headers: List[str]) -> Tuple[bool, str]:
    if not isinstance(record, dict):
        return False, '记录必须是对象'
    normalized = normalize_record(record)
    cells = normalized.get('cells', {})
    if not cells or not isinstance(cells, dict):
        return False, '记录必须包含非空 cells 对象'
    return True, ''


def build_create_records_payload(base_id: str, table_id: str, records: List[Dict[str, Any]]) -> Dict[str, Any]:
    return {
        'baseId': base_id,
        'tableId': table_id,
        'records': [normalize_record(record) for record in records],
    }


def run_mcporter(args: List[str]) -> Optional[Dict[str, Any]]:
    if not args:
        print('错误：空命令')
        return None
    cmd = build_mcporter_call(args)
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if result.returncode != 0:
            print(f"错误：{result.stderr.strip()}")
            return None
        try:
            return json.loads(result.stdout)
        except json.JSONDecodeError as e:
            print(f"无法解析响应：{result.stdout[:200]}...")
            print(f"JSON 解析错误：{e}")
            return None
    except subprocess.TimeoutExpired:
        print('错误：命令执行超时（120 秒）')
        return None
    except FileNotFoundError:
        print('错误：未找到 mcporter 命令，请确认已安装')
        return None


def import_from_csv(base_id: str, table_id: str, csv_file: str, batch_size: int = DEFAULT_BATCH_SIZE) -> bool:
    try:
        safe_path = resolve_safe_path(csv_file)
    except ValueError as e:
        print(f"路径验证失败：{e}")
        return False

    if not validate_file_extension(csv_file, ALLOWED_CSV_EXTENSIONS):
        print(f"错误：只允许 {', '.join(ALLOWED_CSV_EXTENSIONS)} 文件")
        return False
    if not safe_path.exists():
        print(f"错误：文件不存在：{safe_path}")
        return False

    try:
        rows = safe_csv_load(safe_path)
    except ValueError as e:
        print(f"错误：{e}")
        return False
    except csv.Error as e:
        print(f"错误：CSV 格式无效：{e}")
        return False

    if not rows:
        print('错误：CSV 文件为空或没有有效数据行')
        return False

    records = [normalize_record(row) for row in rows if normalize_record(row)['cells']]
    return import_records(base_id, table_id, records, batch_size)


def import_from_json(base_id: str, table_id: str, json_file: str, batch_size: int = DEFAULT_BATCH_SIZE) -> bool:
    try:
        safe_path = resolve_safe_path(json_file)
    except ValueError as e:
        print(f"路径验证失败：{e}")
        return False

    if not validate_file_extension(json_file, ALLOWED_JSON_EXTENSIONS):
        print(f"错误：只允许 {', '.join(ALLOWED_JSON_EXTENSIONS)} 文件")
        return False
    if not safe_path.exists():
        print(f"错误：文件不存在：{safe_path}")
        return False

    try:
        records = safe_json_load(safe_path)
    except ValueError as e:
        print(f"错误：{e}")
        return False
    except json.JSONDecodeError as e:
        print(f"错误：JSON 格式无效：{e}")
        return False

    if not isinstance(records, list) or not records:
        print('错误：JSON 文件必须是非空数组')
        return False

    for i, record in enumerate(records):
        valid, error = validate_record(record, [])
        if not valid:
            print(f"错误：记录 #{i+1} 格式无效：{error}")
            return False

    return import_records(base_id, table_id, [normalize_record(r) for r in records], batch_size)


def import_records(base_id: str, table_id: str, records: List[Dict[str, Any]], batch_size: int) -> bool:
    if batch_size <= 0:
        print('错误：batch_size 必须大于 0')
        return False
    if batch_size > MAX_RECORDS_PER_BATCH:
        batch_size = MAX_RECORDS_PER_BATCH

    total_batches = (len(records) + batch_size - 1) // batch_size
    success = True

    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        batch_num = (i // batch_size) + 1
        payload = build_create_records_payload(base_id, table_id, batch)
        result = run_mcporter(['create_records', '--args', json.dumps(payload, ensure_ascii=False)])
        if result:
            print(f"[{batch_num}/{total_batches}] ✓ 已提交 {len(batch)} 条记录")
        else:
            print(f"[{batch_num}/{total_batches}] ✗ 导入失败")
            success = False

    return success


def main():
    if len(sys.argv) < 4 or len(sys.argv) > 5:
        print(__doc__)
        print('用法示例:')
        print('  python import_records.py basexxx tablexxx data.csv 50')
        sys.exit(1)

    base_id = sys.argv[1]
    table_id = sys.argv[2]
    input_file = sys.argv[3]
    batch_size = int(sys.argv[4]) if len(sys.argv) == 5 else DEFAULT_BATCH_SIZE

    if not validate_resource_id(base_id):
        print('错误：无效的 baseId 格式')
        sys.exit(1)
    if not validate_resource_id(table_id):
        print('错误：无效的 tableId 格式')
        sys.exit(1)

    if input_file.lower().endswith('.csv'):
        success = import_from_csv(base_id, table_id, input_file, batch_size)
    elif input_file.lower().endswith('.json'):
        success = import_from_json(base_id, table_id, input_file, batch_size)
    else:
        print('错误：仅支持 .csv 或 .json 文件')
        sys.exit(1)

    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
新 MCP schema 下的安全与构造测试
"""

import sys
import os
import json
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / 'scripts'))

import bulk_add_fields
import import_records


class TestResolveSafePath(unittest.TestCase):
    def setUp(self):
        self.test_dir = Path(tempfile.mkdtemp())
        self.allowed_file = self.test_dir / 'allowed.json'
        self.allowed_file.write_text('[]')
        self.sub_dir = self.test_dir / 'subdir'
        self.sub_dir.mkdir()
        self.sub_file = self.sub_dir / 'data.csv'
        self.sub_file.write_text('a,b\n1,2')

    def tearDown(self):
        import shutil
        shutil.rmtree(self.test_dir, ignore_errors=True)

    def test_relative_path_within_root(self):
        original_cwd = os.getcwd()
        try:
            os.chdir(self.test_dir)
            result = bulk_add_fields.resolve_safe_path('allowed.json', str(self.test_dir))
            self.assertEqual(result, self.allowed_file.resolve())
        finally:
            os.chdir(original_cwd)

    def test_subdirectory_path(self):
        original_cwd = os.getcwd()
        try:
            os.chdir(self.test_dir)
            result = bulk_add_fields.resolve_safe_path('subdir/data.csv', str(self.test_dir))
            self.assertEqual(result, self.sub_file.resolve())
        finally:
            os.chdir(original_cwd)

    def test_absolute_path_within_root(self):
        result = bulk_add_fields.resolve_safe_path(str(self.allowed_file), str(self.test_dir))
        self.assertEqual(result, self.allowed_file.resolve())

    def test_path_traversal_attack(self):
        with self.assertRaises(ValueError):
            bulk_add_fields.resolve_safe_path('../etc/passwd', str(self.test_dir))

    def test_absolute_path_outside_root(self):
        with self.assertRaises(ValueError):
            bulk_add_fields.resolve_safe_path('/etc/passwd', str(self.test_dir))


class TestResourceIdValidation(unittest.TestCase):
    def test_valid_resource_id(self):
        valid_ids = [
            '123e4567-e89b-12d3-a456-426614174000',
            'base_example_id_12345678',
            'tblABC123_-xyz789',
            'fld_example_12345678\n',
        ]
        for resource_id in valid_ids:
            with self.subTest(resource_id=resource_id):
                self.assertTrue(bulk_add_fields.validate_resource_id(resource_id))
                self.assertTrue(import_records.validate_resource_id(resource_id))

    def test_invalid_resource_id(self):
        invalid_ids = ['', 'short', '含中文', 'has space', 'bad/char', 'a' * 129]
        for resource_id in invalid_ids:
            with self.subTest(resource_id=resource_id):
                self.assertFalse(bulk_add_fields.validate_resource_id(resource_id))
                self.assertFalse(import_records.validate_resource_id(resource_id))


class TestFileExtensionValidation(unittest.TestCase):
    def test_allowed_extensions(self):
        self.assertTrue(bulk_add_fields.validate_file_extension('test.json', ['.json']))
        self.assertTrue(import_records.validate_file_extension('test.csv', ['.csv']))
        self.assertTrue(import_records.validate_file_extension('test.JSON', ['.json']))

    def test_disallowed_extensions(self):
        self.assertFalse(bulk_add_fields.validate_file_extension('test.txt', ['.json']))
        self.assertFalse(import_records.validate_file_extension('test.exe', ['.csv']))


class TestSafeJsonLoad(unittest.TestCase):
    def setUp(self):
        self.test_dir = Path(tempfile.mkdtemp())

    def tearDown(self):
        import shutil
        shutil.rmtree(self.test_dir, ignore_errors=True)

    def test_valid_json(self):
        test_file = self.test_dir / 'valid.json'
        data = [{'fieldName': 'test', 'type': 'text'}]
        test_file.write_text(json.dumps(data))
        self.assertEqual(bulk_add_fields.safe_json_load(test_file), data)

    def test_invalid_json(self):
        test_file = self.test_dir / 'invalid.json'
        test_file.write_text('{invalid json}')
        with self.assertRaises(json.JSONDecodeError):
            bulk_add_fields.safe_json_load(test_file)


class TestFieldConfigValidation(unittest.TestCase):
    def test_valid_field_configs(self):
        valid_configs = [
            {'fieldName': '姓名', 'type': 'text'},
            {'name': '数量', 'type': 'number'},
            {'fieldName': '状态', 'type': 'singleSelect', 'config': {'options': [{'name': '高'}]}},
            {'fieldName': '电话', 'type': 'phone'},
            {'fieldName': '负责人', 'type': 'user', 'config': {'multiple': False}},
        ]
        for config in valid_configs:
            valid, error = bulk_add_fields.validate_field_config(config)
            self.assertTrue(valid, f'{config} should be valid: {error}')

    def test_invalid_field_configs(self):
        invalid_configs = [
            {'type': 'text'},
            {'fieldName': ''},
            {'fieldName': '状态', 'type': 'singleSelect'},
            {'fieldName': '关联', 'type': 'bidirectionalLink', 'config': {}},
            {'fieldName': 'X', 'type': 'invalid_type'},
        ]
        for config in invalid_configs:
            valid, _ = bulk_add_fields.validate_field_config(config)
            self.assertFalse(valid)


class TestRecordValidation(unittest.TestCase):
    def test_valid_record(self):
        self.assertTrue(import_records.validate_record({'cells': {'fldName': '张三'}}, [])[0])
        self.assertTrue(import_records.validate_record({'fldName': '张三'}, [])[0])

    def test_invalid_record(self):
        self.assertFalse(import_records.validate_record({}, [])[0])
        self.assertFalse(import_records.validate_record({'cells': {}}, [])[0])
        self.assertFalse(import_records.validate_record('bad', [])[0])


class TestSanitizeRecordValue(unittest.TestCase):
    def test_string_and_number(self):
        self.assertEqual(import_records.sanitize_record_value('hello'), 'hello')
        self.assertEqual(import_records.sanitize_record_value('123'), 123)
        self.assertEqual(import_records.sanitize_record_value('123.45'), 123.45)

    def test_bool_and_empty(self):
        self.assertIs(import_records.sanitize_record_value('true'), True)
        self.assertIs(import_records.sanitize_record_value('false'), False)
        self.assertIsNone(import_records.sanitize_record_value('   '))
        self.assertIsNone(import_records.sanitize_record_value(None))


class TestPayloadBuilders(unittest.TestCase):
    def test_build_create_fields_payload(self):
        payload = bulk_add_fields.build_create_fields_payload('base12345', 'table12345', [
            {'name': '电话', 'type': 'phone'},
            {'fieldName': '状态', 'type': 'singleSelect', 'config': {'options': [{'name': '高'}]}}
        ])
        self.assertEqual(payload['baseId'], 'base12345')
        self.assertEqual(payload['tableId'], 'table12345')
        self.assertEqual(payload['fields'][0]['fieldName'], '电话')
        self.assertEqual(payload['fields'][0]['type'], 'telephone')

    def test_build_create_records_payload(self):
        payload = import_records.build_create_records_payload('base12345', 'table12345', [
            {'cells': {'fldName': '张三', 'fldAge': '25'}},
            {'fldName': '李四', 'fldActive': 'true'}
        ])
        self.assertEqual(payload['baseId'], 'base12345')
        self.assertEqual(payload['tableId'], 'table12345')
        self.assertEqual(payload['records'][0]['cells']['fldAge'], 25)
        self.assertEqual(payload['records'][1]['cells']['fldActive'], True)


class TestIntegration(unittest.TestCase):
    def setUp(self):
        self.test_dir = Path(tempfile.mkdtemp())
        os.environ['OPENCLAW_WORKSPACE'] = str(self.test_dir)

    def tearDown(self):
        import shutil
        shutil.rmtree(self.test_dir, ignore_errors=True)
        os.environ.pop('OPENCLAW_WORKSPACE', None)

    def test_bulk_add_fields_workflow(self):
        fields_file = self.test_dir / 'fields.json'
        fields = [
            {'fieldName': '任务名', 'type': 'text'},
            {'fieldName': '优先级', 'type': 'singleSelect', 'config': {'options': [{'name': '高'}]}},
        ]
        fields_file.write_text(json.dumps(fields), encoding='utf-8')
        loaded = bulk_add_fields.safe_json_load(fields_file)
        payload = bulk_add_fields.build_create_fields_payload('base12345', 'table12345', loaded)
        self.assertEqual(len(payload['fields']), 2)

    def test_import_records_workflow(self):
        csv_file = self.test_dir / 'data.csv'
        csv_file.write_text('fldName,fldAge\nzhangsan,25\nlisi,30', encoding='utf-8')
        rows = import_records.safe_csv_load(csv_file)
        payload = import_records.build_create_records_payload('base12345', 'table12345', rows)
        self.assertEqual(len(payload['records']), 2)
        self.assertEqual(payload['records'][0]['cells']['fldAge'], 25)


if __name__ == '__main__':
    unittest.main(verbosity=2)

#!/usr/bin/env python3
"""
触发测试：验证技能在正确的场景下被触发
"""

import unittest

class TestSkillTriggering(unittest.TestCase):
    """测试技能触发条件"""

    # 应该触发技能的查询
    SHOULD_TRIGGER = [
        "帮我操作钉钉 AI 表格",
        "创建一个新的 Base",
        "批量导入记录到钉钉表格",
        "查询 AI 表格的记录",
        "更新多维表的数据",
        "添加字段到钉钉表格",
        "从 CSV 导入数据",
        "搜索我的 Base",
        "删除表格中的记录",
        "获取表格结构",
    ]

    # 不应该触发技能的查询
    SHOULD_NOT_TRIGGER = [
        "今天天气怎么样",
        "帮我写 Python 代码",
        "创建 Excel 文件",
        "发送钉钉消息",
        "查询数据库",
        "生成 PDF 报告",
        "翻译这段文字",
        "总结这篇文章",
    ]

    def test_should_trigger_queries(self):
        """测试应该触发的查询"""
        for query in self.SHOULD_TRIGGER:
            with self.subTest(query=query):
                # 这里只是记录测试用例
                # 实际触发测试需要在 Claude 环境中进行
                self.assertIsNotNone(query)

    def test_should_not_trigger_queries(self):
        """测试不应该触发的查询"""
        for query in self.SHOULD_NOT_TRIGGER:
            with self.subTest(query=query):
                self.assertIsNotNone(query)

if __name__ == '__main__':
    print("触发测试用例列表")
    print("\n✅ 应该触发的查询：")
    for i, query in enumerate(TestSkillTriggering.SHOULD_TRIGGER, 1):
        print(f"  {i}. {query}")

    print("\n❌ 不应该触发的查询：")
    for i, query in enumerate(TestSkillTriggering.SHOULD_NOT_TRIGGER, 1):
        print(f"  {i}. {query}")

    print("\n运行单元测试...")
    unittest.main(verbosity=2)