274 lines
9.1 KiB
Python
274 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
批量添加字段到钉钉 AI 表格数据表（新版 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 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', 'address', 'primaryDoc', 'formula',
|
||
'unidirectionalLink', 'bidirectionalLink', 'lookup', 'filterUp',
|
||
'creator', 'lastModifier', 'createdTime', 'lastModifiedTime',
|
||
}
|
||
FIELD_TYPE_ALIASES = {
|
||
'phone': 'telephone',
|
||
}
|
||
|
||
|
||
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_file_extension(filename: str, allowed_extensions: list) -> bool:
|
||
return any(filename.lower().endswith(ext) for ext in allowed_extensions)
|
||
|
||
|
||
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_table_id = (config or {}).get('linkedTableId')
|
||
if not linked_table_id or not validate_resource_id(linked_table_id):
|
||
return False, (
|
||
'关联字段必须提供合法的 config.linkedTableId（目标 Table ID）'
|
||
)
|
||
|
||
if field_type == 'lookup':
|
||
cfg = config or {}
|
||
if not cfg.get('associateField'):
|
||
return False, 'lookup 必须提供 config.associateField（本表关联字段的 fieldId）'
|
||
if not cfg.get('valuesField'):
|
||
return False, 'lookup 必须提供 config.valuesField（关联目标表中要取值的字段 fieldId）'
|
||
if not cfg.get('aggregator'):
|
||
return False, 'lookup 必须提供 config.aggregator（SUM/AVERAGE/COUNT/MAX/MIN/CONCATENATE）'
|
||
|
||
if field_type == 'filterUp':
|
||
cfg = config or {}
|
||
if not cfg.get('targetSheet'):
|
||
return False, 'filterUp 必须提供 config.targetSheet（目标 Table ID）'
|
||
filters = cfg.get('filters')
|
||
if not filters or not isinstance(filters, list):
|
||
return False, 'filterUp 必须提供 config.filters（至少一条筛选规则）'
|
||
if not cfg.get('valuesField'):
|
||
return False, 'filterUp 必须提供 config.valuesField（目标表中要取值的字段 fieldId）'
|
||
if not cfg.get('aggregator'):
|
||
return False, 'filterUp 必须提供 config.aggregator（SUM/AVERAGE/COUNT/MAX/MIN/CONCATENATE）'
|
||
|
||
return True, ''
|
||
|
||
|
||
def build_fields_json(fields: List[Dict[str, Any]]) -> str:
|
||
"""构建 --fields 参数的 JSON 字符串。"""
|
||
payload_fields = []
|
||
for field in fields:
|
||
normalized = normalize_field_config(field)
|
||
item: Dict[str, Any] = {
|
||
'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 json.dumps(payload_fields, ensure_ascii=False)
|
||
|
||
|
||
def run_dws(args: List[str]) -> Optional[Dict[str, Any]]:
|
||
if not args:
|
||
print('错误：空命令')
|
||
return None
|
||
|
||
cmd = ['dws'] + 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('错误：未找到 dws 命令，请确认已安装')
|
||
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
|
||
|
||
fields_json = build_fields_json(fields)
|
||
result = run_dws([
|
||
'aitable', 'field', 'create',
|
||
'--base-id', base_id,
|
||
'--table-id', table_id,
|
||
'--fields', fields_json,
|
||
'--format', 'json',
|
||
])
|
||
|
||
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()
|